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

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.TaskPropertiesHolder;
 
 class LookupListeningAnnouncerConfig
 {
   public static final String DEFAULT_TIER = "__default";
-  private final DataSourceTaskIdHolder dataSourceTaskIdHolder;
+  private final LoadSpecHolder loadSpecHolder;
+  private final TaskPropertiesHolder taskPropsHolder;
+
   @JsonProperty("lookupTier")
   private String lookupTier = null;
   @JsonProperty("lookupTierIsDatasource")
   private boolean lookupTierIsDatasource = false;
 
   @JsonCreator
   public LookupListeningAnnouncerConfig(
-      @JacksonInject DataSourceTaskIdHolder dataSourceTaskIdHolder
+      @JacksonInject LoadSpecHolder loadSpecHolder,
+      @JacksonInject TaskPropertiesHolder taskPropsHolder
   )
   {
-    this.dataSourceTaskIdHolder = dataSourceTaskIdHolder;
+    this.loadSpecHolder = loadSpecHolder;
+    this.taskPropsHolder = taskPropsHolder;
   }
 
   public String getLookupTier()
@@ -50,7 +55,7 @@
         !(lookupTierIsDatasource && null != lookupTier),
         "Cannot specify both `lookupTier` and `lookupTierIsDatasource`"
     );
-    final String lookupTier = lookupTierIsDatasource ? dataSourceTaskIdHolder.getDataSource() : this.lookupTier;
+    final String lookupTier = lookupTierIsDatasource ? taskPropsHolder.getDataSource() : this.lookupTier;
 
     return Preconditions.checkNotNull(
         lookupTier == null ? DEFAULT_TIER : StringUtils.emptyToNullNonDruidDataString(lookupTier),
@@ -61,6 +66,6 @@
 
   public LookupLoadingSpec getLookupLoadingSpec()
   {
-    return dataSourceTaskIdHolder.getLookupLoadingSpec();
+    return loadSpecHolder.getLookupLoadingSpec();
   }
 }

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

@@ -42,7 +42,7 @@
 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 org.apache.druid.server.metrics.LoadSpecHolder;
 
 import javax.annotation.Nullable;
 import java.io.File;
@@ -380,7 +380,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 LoadSpecHolder#getLookupLoadingSpec()}.
    */
   private void loadLookupsAndInitStateRef()
   {

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

@@ -25,7 +25,7 @@
 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.TaskPropertiesHolder;
 
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
@@ -42,10 +42,10 @@ public class ChatHandlerResource
   private final String taskId;
 
   @Inject
-  public ChatHandlerResource(final ChatHandlerProvider handlers, final DataSourceTaskIdHolder taskIdHolder)
+  public ChatHandlerResource(final ChatHandlerProvider handlers, final TaskPropertiesHolder taskPropsHolder)
   {
     this.handlers = handlers;
-    this.taskId = taskIdHolder.getTaskId();
+    this.taskId = taskPropsHolder.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;
@@ -82,7 +82,7 @@ public class SegmentCacheBootstrapper
 
   private static final EmittingLogger log = new EmittingLogger(SegmentCacheBootstrapper.class);
 
-  private final DataSourceTaskIdHolder datasourceHolder;
+  private final LoadSpecHolder loadSpecHolder;
 
   @Inject
   public SegmentCacheBootstrapper(
@@ -94,7 +94,7 @@ public SegmentCacheBootstrapper(
       ServerTypeConfig serverTypeConfig,
       CoordinatorClient coordinatorClient,
       ServiceEmitter emitter,
-      DataSourceTaskIdHolder datasourceHolder
+      LoadSpecHolder loadSpecHolder
   )
   {
     this.loadDropHandler = loadDropHandler;
@@ -105,7 +105,7 @@ public SegmentCacheBootstrapper(
     this.serverTypeConfig = serverTypeConfig;
     this.coordinatorClient = coordinatorClient;
     this.emitter = emitter;
-    this.datasourceHolder = datasourceHolder;
+    this.loadSpecHolder = loadSpecHolder;
   }
 
   @LifecycleStart
@@ -270,12 +270,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()}
+   * 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 +290,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.TaskPropertiesHolder;
 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 TaskPropertiesHolder taskPropsHolder
   )
   {
-    return new TaskIdResponseHeaderFilterHolder("/druid/worker/v1/chat/*", taskIdHolder.getTaskId());
+    return new TaskIdResponseHeaderFilterHolder("/druid/worker/v1/chat/*", taskPropsHolder.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.TaskPropertiesHolder;
 import org.apache.druid.server.security.TLSCertificateChecker;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
@@ -109,7 +109,7 @@ public void configure(Binder binder)
   @Provides
   @LazySingleton
   public TaskIdResponseHeaderFilterHolder taskIdResponseHeaderFilterHolderBuilder(
-      final DataSourceTaskIdHolder taskIdHolder
+      final TaskPropertiesHolder taskIdHolder
   )
   {
     return new TaskIdResponseHeaderFilterHolder("/druid/worker/v1/chat/*", taskIdHolder.getTaskId());

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.TaskPropertiesHolder;
 import org.apache.druid.server.security.CustomCheckX509TrustManager;
 import org.apache.druid.server.security.TLSCertificateChecker;
 import org.eclipse.jetty.server.ConnectionFactory;
@@ -517,11 +517,11 @@ private static int getTCPAcceptQueueSize()
 
   @Provides
   @LazySingleton
-  public JettyMonitor getJettyMonitor(DataSourceTaskIdHolder dataSourceTaskIdHolder)
+  public JettyMonitor getJettyMonitor(TaskPropertiesHolder taskPropsHolder)
   {
     return new JettyMonitor(
         MonitorsConfig.mapOfDatasourceAndTaskID(
-            dataSourceTaskIdHolder.getDataSource(), dataSourceTaskIdHolder.getTaskId()
+            taskPropsHolder.getDataSource(), taskPropsHolder.getTaskId()
         )
     );
   }

server/src/main/java/org/apache/druid/server/metrics/GroupByStatsMonitor.java

@@ -49,14 +49,14 @@ public class GroupByStatsMonitor extends AbstractMonitor
   public GroupByStatsMonitor(
       GroupByStatsProvider groupByStatsProvider,
       @Merging BlockingPool<ByteBuffer> mergeBufferPool,
-      DataSourceTaskIdHolder dataSourceTaskIdHolder
+      TaskPropertiesHolder taskPropsHolder
   )
   {
     this.groupByStatsProvider = groupByStatsProvider;
     this.mergeBufferPool = mergeBufferPool;
     this.dimensions = MonitorsConfig.mapOfDatasourceAndTaskID(
-        dataSourceTaskIdHolder.getDataSource(),
-        dataSourceTaskIdHolder.getTaskId()
+        taskPropsHolder.getDataSource(),
+        taskPropsHolder.getTaskId()
     );
   }
 

server/src/main/java/org/apache/druid/server/metrics/DataSourceTaskIdHolder.java → server/src/main/java/org/apache/druid/server/metrics/LoadSpecHolder.java

@@ -24,26 +24,15 @@
 import org.apache.druid.server.coordination.BroadcastDatasourceLoadingSpec;
 import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
 
-import javax.annotation.Nullable;
-
 /**
- * This holder is only applicable to {@code CliPeon} servers.
+ * A holder applicable to all servers that contains load specifications such as {@link LookupLoadingSpec}
+ * and {@link BroadcastDatasourceLoadingSpec}.
  */
-public class DataSourceTaskIdHolder
+public class LoadSpecHolder
 {
-  public static final String DATA_SOURCE_BINDING = "druidDataSource";
-  public static final String TASK_ID_BINDING = "druidTaskId";
   public static final String LOOKUPS_TO_LOAD_FOR_TASK = "lookupsToLoadForTask";
   public static final String BROADCAST_DATASOURCES_TO_LOAD_FOR_TASK = "broadcastDatasourcesToLoadForTask";
 
-  @Named(DATA_SOURCE_BINDING)
-  @Inject(optional = true)
-  String dataSource = null;
-
-  @Named(TASK_ID_BINDING)
-  @Inject(optional = true)
-  String taskId = null;
-
   @Named(LOOKUPS_TO_LOAD_FOR_TASK)
   @Inject(optional = true)
   LookupLoadingSpec lookupLoadingSpec = LookupLoadingSpec.ALL;
@@ -52,24 +41,6 @@ public class DataSourceTaskIdHolder
   @Inject(optional = true)
   BroadcastDatasourceLoadingSpec broadcastDatasourceLoadingSpec = BroadcastDatasourceLoadingSpec.ALL;
 
-  /**
-   * @return the taskId for CliPeon servers; {@code null} for all other servers.
-   */
-  @Nullable
-  public String getDataSource()
-  {
-    return dataSource;
-  }
-
-  /**
-   * @return the dataSource for CliPeon servers; {@code null} for all other servers.
-   */
-  @Nullable
-  public String getTaskId()
-  {
-    return taskId;
-  }
-
   public LookupLoadingSpec getLookupLoadingSpec()
   {
     return lookupLoadingSpec;

server/src/main/java/org/apache/druid/server/metrics/MetricsModule.java

@@ -87,7 +87,7 @@ public void configure(Binder binder)
 
     DruidBinders.metricMonitorBinder(binder); // get the binder so that it will inject the empty set at a minimum.
 
-    binder.bind(DataSourceTaskIdHolder.class).in(LazySingleton.class);
+    binder.bind(TaskPropertiesHolder.class).in(LazySingleton.class);
 
     binder.bind(ExecutorServiceMonitor.class).in(LazySingleton.class);
 
@@ -111,8 +111,8 @@ public MonitorScheduler getMonitorScheduler(
     List<Monitor> monitors = new ArrayList<>();
     // HACK: when ServiceStatusMonitor is the first to be loaded, it introduces a circular dependency between
     // CliPeon.runTask and CliPeon.getDataSourceFromTask/CliPeon.getTaskIDFromTask. The reason for this is unclear
-    // but by injecting DataSourceTaskIdHolder early this cycle is avoided.
-    injector.getInstance(DataSourceTaskIdHolder.class);
+    // but by injecting TaskPropertiesHolder early this cycle is avoided.
+    injector.getInstance(TaskPropertiesHolder.class);
     for (Class<? extends Monitor> monitorClass : Iterables.concat(monitorsConfig.getMonitors(), monitorSet)) {
       if (shouldLoadMonitor(monitorClass, nodeRoles)) {
         monitors.add(injector.getInstance(monitorClass));
@@ -150,50 +150,50 @@ public MonitorScheduler getMonitorScheduler(
   @Provides
   @ManageLifecycle
   public JvmMonitor getJvmMonitor(
-      DataSourceTaskIdHolder dataSourceTaskIdHolder
+      TaskPropertiesHolder taskPropsHolder
   )
   {
     Map<String, String[]> dimensions = MonitorsConfig.mapOfDatasourceAndTaskID(
-        dataSourceTaskIdHolder.getDataSource(),
-        dataSourceTaskIdHolder.getTaskId()
+        taskPropsHolder.getDataSource(),
+        taskPropsHolder.getTaskId()
     );
     return new JvmMonitor(dimensions);
   }
 
   @Provides
   @ManageLifecycle
   public JvmCpuMonitor getJvmCpuMonitor(
-      DataSourceTaskIdHolder dataSourceTaskIdHolder
+      TaskPropertiesHolder taskPropsHolder
   )
   {
     Map<String, String[]> dimensions = MonitorsConfig.mapOfDatasourceAndTaskID(
-        dataSourceTaskIdHolder.getDataSource(),
-        dataSourceTaskIdHolder.getTaskId()
+        taskPropsHolder.getDataSource(),
+        taskPropsHolder.getTaskId()
     );
     return new JvmCpuMonitor(dimensions);
   }
 
   @Provides
   @ManageLifecycle
-  public JvmThreadsMonitor getJvmThreadsMonitor(DataSourceTaskIdHolder dataSourceTaskIdHolder)
+  public JvmThreadsMonitor getJvmThreadsMonitor(TaskPropertiesHolder taskPropsHolder)
   {
     Map<String, String[]> dimensions = MonitorsConfig.mapOfDatasourceAndTaskID(
-        dataSourceTaskIdHolder.getDataSource(),
-        dataSourceTaskIdHolder.getTaskId()
+        taskPropsHolder.getDataSource(),
+        taskPropsHolder.getTaskId()
     );
     return new JvmThreadsMonitor(dimensions);
   }
 
   @Provides
   @ManageLifecycle
-  public SysMonitor getSysMonitor(DataSourceTaskIdHolder dataSourceTaskIdHolder, @Self Set<NodeRole> nodeRoles)
+  public SysMonitor getSysMonitor(TaskPropertiesHolder taskPropsHolder, @Self Set<NodeRole> nodeRoles)
   {
     if (nodeRoles.contains(NodeRole.PEON)) {
       return new NoopSysMonitor();
     } else {
       Map<String, String[]> dimensions = MonitorsConfig.mapOfDatasourceAndTaskID(
-          dataSourceTaskIdHolder.getDataSource(),
-          dataSourceTaskIdHolder.getTaskId()
+          taskPropsHolder.getDataSource(),
+          taskPropsHolder.getTaskId()
       );
       return new SysMonitor(dimensions);
     }
@@ -202,7 +202,7 @@ public SysMonitor getSysMonitor(DataSourceTaskIdHolder dataSourceTaskIdHolder, @
   @Provides
   @ManageLifecycle
   public OshiSysMonitor getOshiSysMonitor(
-      DataSourceTaskIdHolder dataSourceTaskIdHolder,
+      TaskPropertiesHolder taskPropsHolder,
       @Self Set<NodeRole> nodeRoles,
       OshiSysMonitorConfig oshiSysConfig
   )
@@ -211,8 +211,8 @@ public OshiSysMonitor getOshiSysMonitor(
       return new NoopOshiSysMonitor();
     } else {
       Map<String, String[]> dimensions = MonitorsConfig.mapOfDatasourceAndTaskID(
-          dataSourceTaskIdHolder.getDataSource(),
-          dataSourceTaskIdHolder.getTaskId()
+          taskPropsHolder.getDataSource(),
+          taskPropsHolder.getTaskId()
       );
       return new OshiSysMonitor(dimensions, oshiSysConfig);
     }