Add "includeAllCounters()" to WorkerContext.

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerContext.java

@@ -27,6 +27,7 @@
 import org.apache.druid.msq.kernel.FrameProcessorFactory;
 import org.apache.druid.msq.kernel.QueryDefinition;
 import org.apache.druid.msq.kernel.WorkOrder;
+import org.apache.druid.msq.util.MultiStageQueryContext;
 import org.apache.druid.server.DruidNode;
 
 import java.io.File;
@@ -98,4 +99,9 @@ public interface WorkerContext
   DruidNode selfNode();
 
   DataServerQueryHandlerFactory dataServerQueryHandlerFactory();
+
+  /**
+   * Whether to include all counters in reports. See {@link MultiStageQueryContext#CTX_INCLUDE_ALL_COUNTERS} for detail.
+   */
+  boolean includeAllCounters();
 }

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java

@@ -407,7 +407,7 @@ private void handleNewWorkOrder(
     kernel.startReading();
 
     final QueryContext queryContext = task != null ? QueryContext.of(task.getContext()) : QueryContext.empty();
-    final boolean includeAllCounters = MultiStageQueryContext.getIncludeAllCounters(queryContext);
+    final boolean includeAllCounters = context.includeAllCounters();
     final RunWorkOrder runWorkOrder = new RunWorkOrder(
         task.getControllerTaskId(),
         workOrder,

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java

@@ -312,6 +312,12 @@ public DataServerQueryHandlerFactory dataServerQueryHandlerFactory()
     return dataServerQueryHandlerFactory;
   }
 
+  @Override
+  public boolean includeAllCounters()
+  {
+    return includeAllCounters;
+  }
+
   private synchronized ServiceLocator makeControllerLocator(final String controllerId)
   {
     if (controllerLocator == null) {

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestWorkerContext.java

@@ -159,6 +159,12 @@ public DataServerQueryHandlerFactory dataServerQueryHandlerFactory()
     return injector.getInstance(DataServerQueryHandlerFactory.class);
   }
 
+  @Override
+  public boolean includeAllCounters()
+  {
+    return true;
+  }
+
   class FrameContextImpl implements FrameContext
   {
     private final File tempDir;