Refactor: Rename DataSourceTaskIdHolder and extract load spec holder to its own class

indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java

@@ -1240,6 +1240,8 @@ public static class Builder
     private final String dataSource;
     private final SegmentCacheManagerFactory segmentCacheManagerFactory;
 
+    @Nullable
+    private String id;
     private CompactionIOConfig ioConfig;
     @Nullable
     private DimensionsSpec dimensionsSpec;
@@ -1268,6 +1270,12 @@ public Builder(
       this.segmentCacheManagerFactory = segmentCacheManagerFactory;
     }
 
+    public Builder id(String id)
+    {
+      this.id = id;
+      return this;
+    }
+
     public Builder interval(Interval interval)
     {
       return inputSpec(new CompactionIntervalSpec(interval, null));
@@ -1353,7 +1361,7 @@ public Builder projections(@Nullable List<AggregateProjectionSpec> projections)
     public CompactionTask build()
     {
       return new CompactionTask(
-          null,
+          id,
           null,
           dataSource,
           null,

quidem-ut/src/main/java/org/apache/druid/quidem/ExposedAsBrokerQueryComponentSupplierWrapper.java

@@ -41,6 +41,7 @@
 import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
 import org.apache.druid.guice.AnnouncerModule;
 import org.apache.druid.guice.BrokerServiceModule;
+import org.apache.druid.guice.DefaultServerHolderModule;
 import org.apache.druid.guice.ExpressionModule;
 import org.apache.druid.guice.ExtensionsModule;
 import org.apache.druid.guice.JacksonConfigManagerModule;
@@ -194,6 +195,7 @@ private List<Module> forServerModules()
         new ExternalStorageAccessSecurityModule(),
         new ServiceClientModule(),
         new StorageConnectorModule(),
+        new DefaultServerHolderModule(),
         ServerInjectorBuilder.registerNodeRoleModule(ImmutableSet.of())
     );
   }

server/src/main/java/org/apache/druid/guice/DefaultServerHolderModule.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.druid.guice;
+
+import com.google.inject.Binder;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.guice.annotations.ExcludeScope;
+import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.server.metrics.DefaultLoadSpecHolder;
+import org.apache.druid.server.metrics.LoadSpecHolder;
+import org.apache.druid.server.metrics.NoopTaskHolder;
+import org.apache.druid.server.metrics.TaskHolder;
+
+/**
+ * Binds the following holder configs for all servers except {@code CliPeon}:
+ * <ul>
+ * <li>{@link TaskHolder} to {@link NoopTaskHolder}</li>
+ * <li>{@link LoadSpecHolder} to {@link DefaultLoadSpecHolder}</li>
+ * </ul>
+ *
+ * <p>For {@code CliPeon}, these bindings are overridden by the peon-specific module.</p>
+ */
+@ExcludeScope(roles = {NodeRole.PEON_JSON_NAME})
+public class DefaultServerHolderModule implements DruidModule
+{
+  @Override
+  public void configure(Binder binder)
+  {
+    binder.bind(TaskHolder.class).to(NoopTaskHolder.class).in(LazySingleton.class);
+    binder.bind(LoadSpecHolder.class).to(DefaultLoadSpecHolder.class).in(LazySingleton.class);
+  }
+}

server/src/main/java/org/apache/druid/guice/annotations/ExcludeScope.java

@@ -25,7 +25,7 @@
 import java.lang.annotation.Target;
 
 /**
- * An annotation to exlcude specific node types that a {@link com.google.inject.Module} can be loaded on.
+ * An annotation to exclude specific node types that a {@link com.google.inject.Module} can be loaded on.
  * The {@link #roles()} should be the {@link org.apache.druid.discovery.NodeRole#jsonName}. If both {@link ExcludeScope}
  * and {@link LoadScope} are set, {@link ExcludeScope} takes precedence.
  * <p>

server/src/main/java/org/apache/druid/initialization/CoreInjectorBuilder.java

@@ -26,6 +26,7 @@
 import org.apache.druid.guice.AnnouncerModule;
 import org.apache.druid.guice.BuiltInTypesModule;
 import org.apache.druid.guice.CatalogCoreModule;
+import org.apache.druid.guice.DefaultServerHolderModule;
 import org.apache.druid.guice.DruidInjectorBuilder;
 import org.apache.druid.guice.DruidSecondaryModule;
 import org.apache.druid.guice.ExpressionModule;
@@ -142,7 +143,8 @@ public CoreInjectorBuilder forServerWithoutJetty()
         new ExternalStorageAccessSecurityModule(),
         new ServiceClientModule(),
         new StorageConnectorModule(),
-        new CatalogCoreModule()
+        new CatalogCoreModule(),
+        new DefaultServerHolderModule()
     );
     return this;
   }

server/src/main/java/org/apache/druid/query/lookup/LookupListeningAnnouncerConfig.java

@@ -25,23 +25,28 @@
 import com.google.common.base.Preconditions;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
-import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
+import org.apache.druid.server.metrics.LoadSpecHolder;
+import org.apache.druid.server.metrics.TaskHolder;
 
 class LookupListeningAnnouncerConfig
 {
   public static final String DEFAULT_TIER = "__default";
-  private final DataSourceTaskIdHolder dataSourceTaskIdHolder;
+  private final TaskHolder taskHolder;
+  private final LoadSpecHolder loadSpecHolder;
+
   @JsonProperty("lookupTier")
   private String lookupTier = null;
   @JsonProperty("lookupTierIsDatasource")
   private boolean lookupTierIsDatasource = false;
 
   @JsonCreator
   public LookupListeningAnnouncerConfig(
-      @JacksonInject DataSourceTaskIdHolder dataSourceTaskIdHolder
+      @JacksonInject TaskHolder taskHolder,
+      @JacksonInject LoadSpecHolder loadSpecHolder
   )
   {
-    this.dataSourceTaskIdHolder = dataSourceTaskIdHolder;
+    this.taskHolder = taskHolder;
+    this.loadSpecHolder = loadSpecHolder;
   }
 
   public String getLookupTier()
@@ -50,7 +55,8 @@
         !(lookupTierIsDatasource && null != lookupTier),
         "Cannot specify both `lookupTier` and `lookupTierIsDatasource`"
     );
-    final String lookupTier = lookupTierIsDatasource ? dataSourceTaskIdHolder.getDataSource() : this.lookupTier;
+
+    final String lookupTier = lookupTierIsDatasource ? taskHolder.getDataSource() : this.lookupTier;
 
     return Preconditions.checkNotNull(
         lookupTier == null ? DEFAULT_TIER : StringUtils.emptyToNullNonDruidDataString(lookupTier),
@@ -61,6 +67,6 @@
 
   public LookupLoadingSpec getLookupLoadingSpec()
   {
-    return dataSourceTaskIdHolder.getLookupLoadingSpec();
+    return loadSpecHolder.getLookupLoadingSpec();
   }
 }

server/src/main/java/org/apache/druid/query/lookup/LookupReferencesManager.java

@@ -42,7 +42,6 @@
 import org.apache.druid.java.util.emitter.EmittingLogger;
 import org.apache.druid.server.http.ServletResourceUtils;
 import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
-import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
 
 import javax.annotation.Nullable;
 import java.io.File;
@@ -380,7 +379,7 @@ private void takeSnapshot(Map<String, LookupExtractorFactoryContainer> lookupMap
   }
 
   /**
-   * Load a set of lookups based on the injected value in {@link DataSourceTaskIdHolder#getLookupLoadingSpec()}.
+   * Load a set of lookups based on the injected value in {@link TaskPropertiesHolder#getLookupLoadingSpec()}.
    */
   private void loadLookupsAndInitStateRef()
   {

server/src/main/java/org/apache/druid/segment/realtime/ChatHandlerResource.java

@@ -25,8 +25,9 @@
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.server.initialization.jetty.BadRequestException;
 import org.apache.druid.server.initialization.jetty.ServiceUnavailableException;
-import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
+import org.apache.druid.server.metrics.TaskHolder;
 
+import javax.annotation.Nullable;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.core.Context;
@@ -39,13 +40,14 @@ public class ChatHandlerResource
   public static final String TASK_ID_HEADER = "X-Druid-Task-Id";
 
   private final ChatHandlerProvider handlers;
+  @Nullable
   private final String taskId;
 
   @Inject
-  public ChatHandlerResource(final ChatHandlerProvider handlers, final DataSourceTaskIdHolder taskIdHolder)
+  public ChatHandlerResource(final ChatHandlerProvider handlers, final TaskHolder taskHolder)
   {
     this.handlers = handlers;
-    this.taskId = taskIdHolder.getTaskId();
+    this.taskId = taskHolder.getTaskId();
   }
 
   @Path("/{id}")

server/src/main/java/org/apache/druid/server/coordination/SegmentCacheBootstrapper.java

@@ -39,7 +39,7 @@
 import org.apache.druid.segment.loading.SegmentLoaderConfig;
 import org.apache.druid.segment.loading.SegmentLoadingException;
 import org.apache.druid.server.SegmentManager;
-import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
+import org.apache.druid.server.metrics.LoadSpecHolder;
 import org.apache.druid.timeline.DataSegment;
 
 import javax.annotation.Nullable;
@@ -74,6 +74,7 @@ public class SegmentCacheBootstrapper
   private final ServerTypeConfig serverTypeConfig;
   private final CoordinatorClient coordinatorClient;
   private final ServiceEmitter emitter;
+  private final LoadSpecHolder loadSpecHolder;
 
   private volatile boolean isComplete = false;
 
@@ -82,8 +83,6 @@ public class SegmentCacheBootstrapper
 
   private static final EmittingLogger log = new EmittingLogger(SegmentCacheBootstrapper.class);
 
-  private final DataSourceTaskIdHolder datasourceHolder;
-
   @Inject
   public SegmentCacheBootstrapper(
       SegmentLoadDropHandler loadDropHandler,
@@ -94,7 +93,7 @@ public SegmentCacheBootstrapper(
       ServerTypeConfig serverTypeConfig,
       CoordinatorClient coordinatorClient,
       ServiceEmitter emitter,
-      DataSourceTaskIdHolder datasourceHolder
+      LoadSpecHolder loadSpecHolder
   )
   {
     this.loadDropHandler = loadDropHandler;
@@ -105,7 +104,7 @@ public SegmentCacheBootstrapper(
     this.serverTypeConfig = serverTypeConfig;
     this.coordinatorClient = coordinatorClient;
     this.emitter = emitter;
-    this.datasourceHolder = datasourceHolder;
+    this.loadSpecHolder = loadSpecHolder;
   }
 
   @LifecycleStart
@@ -270,12 +269,12 @@ private void loadSegmentsOnStartup() throws IOException
 
   /**
    * @return a list of bootstrap segments. When bootstrap segments cannot be found, an empty list is returned.
-   * The bootstrap segments returned are filtered by the broadcast datasources indicated by {@link DataSourceTaskIdHolder#getBroadcastDatasourceLoadingSpec()}
-   * if applicable.
+   * The bootstrap segments returned are filtered by the broadcast datasources indicated by
+   * {@link LoadSpecHolder#getBroadcastDatasourceLoadingSpec()} if applicable.
    */
   private List<DataSegment> getBootstrapSegments()
   {
-    final BroadcastDatasourceLoadingSpec.Mode mode = datasourceHolder.getBroadcastDatasourceLoadingSpec().getMode();
+    final BroadcastDatasourceLoadingSpec.Mode mode = loadSpecHolder.getBroadcastDatasourceLoadingSpec().getMode();
     if (mode == BroadcastDatasourceLoadingSpec.Mode.NONE) {
       log.info("Skipping fetch of bootstrap segments.");
       return ImmutableList.of();
@@ -290,7 +289,7 @@ private List<DataSegment> getBootstrapSegments()
       final BootstrapSegmentsResponse response =
           FutureUtils.getUnchecked(coordinatorClient.fetchBootstrapSegments(), true);
       if (mode == BroadcastDatasourceLoadingSpec.Mode.ONLY_REQUIRED) {
-        final Set<String> broadcastDatasourcesToLoad = datasourceHolder.getBroadcastDatasourceLoadingSpec().getBroadcastDatasourcesToLoad();
+        final Set<String> broadcastDatasourcesToLoad = loadSpecHolder.getBroadcastDatasourceLoadingSpec().getBroadcastDatasourcesToLoad();
         final List<DataSegment> filteredBroadcast = new ArrayList<>();
         response.getIterator().forEachRemaining(segment -> {
           if (broadcastDatasourcesToLoad.contains(segment.getDataSource())) {

server/src/main/java/org/apache/druid/server/initialization/jetty/ChatHandlerServerModule.java

@@ -35,7 +35,7 @@
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.initialization.ServerConfig;
 import org.apache.druid.server.initialization.TLSServerConfig;
-import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
+import org.apache.druid.server.metrics.TaskHolder;
 import org.apache.druid.server.security.TLSCertificateChecker;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
@@ -83,10 +83,10 @@ public void configure(Binder binder)
   @Provides
   @LazySingleton
   public TaskIdResponseHeaderFilterHolder taskIdResponseHeaderFilterHolderBuilder(
-      final DataSourceTaskIdHolder taskIdHolder
+      final TaskHolder taskHolder
   )
   {
-    return new TaskIdResponseHeaderFilterHolder("/druid/worker/v1/chat/*", taskIdHolder.getTaskId());
+    return new TaskIdResponseHeaderFilterHolder("/druid/worker/v1/chat/*", taskHolder.getTaskId());
   }
 
   @Provides

server/src/main/java/org/apache/druid/server/initialization/jetty/CliIndexerServerModule.java

@@ -36,7 +36,7 @@
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.initialization.ServerConfig;
 import org.apache.druid.server.initialization.TLSServerConfig;
-import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
+import org.apache.druid.server.metrics.TaskHolder;
 import org.apache.druid.server.security.TLSCertificateChecker;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
@@ -109,10 +109,10 @@ public void configure(Binder binder)
   @Provides
   @LazySingleton
   public TaskIdResponseHeaderFilterHolder taskIdResponseHeaderFilterHolderBuilder(
-      final DataSourceTaskIdHolder taskIdHolder
+      final TaskHolder taskHolder
   )
   {
-    return new TaskIdResponseHeaderFilterHolder("/druid/worker/v1/chat/*", taskIdHolder.getTaskId());
+    return new TaskIdResponseHeaderFilterHolder("/druid/worker/v1/chat/*", taskHolder.getTaskId());
   }
 
   @Provides

server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java

@@ -53,9 +53,9 @@
 import org.apache.druid.server.StatusResource;
 import org.apache.druid.server.initialization.ServerConfig;
 import org.apache.druid.server.initialization.TLSServerConfig;
-import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
 import org.apache.druid.server.metrics.MetricsModule;
 import org.apache.druid.server.metrics.MonitorsConfig;
+import org.apache.druid.server.metrics.TaskHolder;
 import org.apache.druid.server.security.CustomCheckX509TrustManager;
 import org.apache.druid.server.security.TLSCertificateChecker;
 import org.eclipse.jetty.server.ConnectionFactory;
@@ -517,12 +517,10 @@ private static int getTCPAcceptQueueSize()
 
   @Provides
   @LazySingleton
-  public JettyMonitor getJettyMonitor(DataSourceTaskIdHolder dataSourceTaskIdHolder)
+  public JettyMonitor getJettyMonitor(TaskHolder taskHolder)
   {
     return new JettyMonitor(
-        MonitorsConfig.mapOfDatasourceAndTaskID(
-            dataSourceTaskIdHolder.getDataSource(), dataSourceTaskIdHolder.getTaskId()
-        )
+        MonitorsConfig.mapOfTaskHolderDimensions(taskHolder)
     );
   }
 

server/src/main/java/org/apache/druid/server/initialization/jetty/TaskIdResponseHeaderFilterHolder.java

@@ -23,9 +23,11 @@
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.segment.realtime.ChatHandlerResource;
 
+import javax.annotation.Nullable;
+
 public class TaskIdResponseHeaderFilterHolder extends ResponseHeaderFilterHolder
 {
-  public TaskIdResponseHeaderFilterHolder(String path, String taskId)
+  public TaskIdResponseHeaderFilterHolder(String path, @Nullable String taskId)
   {
     super(path,
           taskId == null