diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Task.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Task.java
index cacdc47f520a..1fadd4f6ae25 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Task.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Task.java
@@ -343,4 +343,14 @@ default BroadcastDatasourceLoadingSpec getBroadcastDatasourceLoadingSpec()
   {
     return BroadcastDatasourceLoadingSpec.createFromContext(getContext(), BroadcastDatasourceLoadingSpec.ALL);
   }
+
+  /**
+   * specifies the current status of the task.
+   *
+   * @return string
+   */
+  default String status()
+  {
+    return "UNKONWN";
+  }
 }
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java
index 41cd084cd960..606ce6334923 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java
@@ -311,4 +311,11 @@ public Appenderator getAppenderator()
   {
     return runnerSupplier.get();
   }
+
+  @Override
+  @VisibleForTesting
+  public String status()
+  {
+    return getRunner().getStatus().name();
+  }
 }
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
index 40bbe84b623a..91b5c30482ce 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
@@ -2492,7 +2492,6 @@ public void testScheduleReporting()
   {
     EasyMock.expect(spec.isSuspended()).andReturn(false).anyTimes();
     DruidMonitorSchedulerConfig config = new DruidMonitorSchedulerConfig();
-    EasyMock.expect(spec.getMonitorSchedulerConfig()).andReturn(config).times(2);
     ScheduledExecutorService executorService = EasyMock.createMock(ScheduledExecutorService.class);
     EasyMock.expect(executorService.scheduleWithFixedDelay(EasyMock.anyObject(), EasyMock.eq(86415000L), EasyMock.eq(300000L), EasyMock.eq(TimeUnit.MILLISECONDS))).andReturn(EasyMock.createMock(ScheduledFuture.class)).once();
     EasyMock.expect(executorService.scheduleAtFixedRate(EasyMock.anyObject(), EasyMock.eq(86425000L), EasyMock.eq(config.getEmissionDuration().getMillis()), EasyMock.eq(TimeUnit.MILLISECONDS))).andReturn(EasyMock.createMock(ScheduledFuture.class)).times(2);
diff --git a/services/src/main/java/org/apache/druid/cli/CliPeon.java b/services/src/main/java/org/apache/druid/cli/CliPeon.java
index 15374625d301..cf0403f58e3d 100644
--- a/services/src/main/java/org/apache/druid/cli/CliPeon.java
+++ b/services/src/main/java/org/apache/druid/cli/CliPeon.java
@@ -25,10 +25,8 @@
 import com.github.rvesse.airline.annotations.Option;
 import com.github.rvesse.airline.annotations.restrictions.Required;
 import com.google.common.base.Supplier;
-import com.google.common.base.Suppliers;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Binder;
 import com.google.inject.Inject;
@@ -141,6 +139,7 @@
 import java.io.IOException;
 import java.nio.charset.Charset;
 import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -303,19 +302,20 @@ public void configure(Binder binder)
           @Named(ServiceStatusMonitor.HEARTBEAT_TAGS_BINDING)
           public Supplier<Map<String, Object>> heartbeatDimensions(Task task)
           {
-            ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
-            builder.put(DruidMetrics.TASK_ID, task.getId());
-            builder.put(DruidMetrics.DATASOURCE, task.getDataSource());
-            builder.put(DruidMetrics.TASK_TYPE, task.getType());
-            builder.put(DruidMetrics.GROUP_ID, task.getGroupId());
+            Map<String, Object> map = new HashMap<>();
+            map.put(DruidMetrics.TASK_ID, task.getId());
+            map.put(DruidMetrics.DATASOURCE, task.getDataSource());
+            map.put(DruidMetrics.TASK_TYPE, task.getType());
+            map.put(DruidMetrics.GROUP_ID, task.getGroupId());
             Map<String, Object> tags = task.getContextValue(DruidMetrics.TAGS);
             if (tags != null && !tags.isEmpty()) {
-              builder.put(DruidMetrics.TAGS, tags);
+              map.put(DruidMetrics.TAGS, tags);
             }
 
-            return Suppliers.ofInstance(
-                builder.build()
-            );
+            return () -> {
+              map.put(DruidMetrics.STATUS, task.status());
+              return map;
+            };
           }
 
           @Provides