Disable ZK-based segment loading, remove usage of skife from DruidCoordinatorConfig

server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java

@@ -54,6 +54,9 @@
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.coordinator.balancer.BalancerStrategyFactory;
 import org.apache.druid.server.coordinator.compact.CompactionSegmentSearchPolicy;
+import org.apache.druid.server.coordinator.config.CoordinatorKillConfigs;
+import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.KillUnusedSegmentsConfig;
 import org.apache.druid.server.coordinator.duty.BalanceSegments;
 import org.apache.druid.server.coordinator.duty.CollectSegmentAndServerStats;
 import org.apache.druid.server.coordinator.duty.CompactSegments;
@@ -182,7 +185,6 @@ public DruidCoordinator(
       ServiceAnnouncer serviceAnnouncer,
       @Self DruidNode self,
       CoordinatorCustomDutyGroups customDutyGroups,
-      BalancerStrategyFactory balancerStrategyFactory,
       LookupCoordinatorManager lookupCoordinatorManager,
       @Coordinator DruidLeaderSelector coordLeaderSelector,
       CompactionSegmentSearchPolicy compactionSegmentSearchPolicy
@@ -200,7 +202,7 @@ public DruidCoordinator(
 
     this.executorFactory = scheduledExecutorFactory;
 
-    this.balancerStrategyFactory = balancerStrategyFactory;
+    this.balancerStrategyFactory = config.getBalancerStrategyFactory();
     this.lookupCoordinatorManager = lookupCoordinatorManager;
     this.coordLeaderSelector = coordLeaderSelector;
     this.compactSegments = initializeCompactSegmentsDuty(compactionSegmentSearchPolicy);
@@ -545,10 +547,13 @@ private List<CoordinatorDuty> makeHistoricalManagementDuties()
   List<CoordinatorDuty> makeIndexingServiceDuties()
   {
     final List<CoordinatorDuty> duties = new ArrayList<>();
-    if (config.isKillUnusedSegmentsEnabled()) {
-      duties.add(new KillUnusedSegments(metadataManager.segments(), overlordClient, config));
+    final KillUnusedSegmentsConfig killUnusedConfig = config.getKillConfigs().unusedSegments(
+        config.getCoordinatorIndexingPeriod()
+    );
+    if (killUnusedConfig.isCleanupEnabled()) {
+      duties.add(new KillUnusedSegments(metadataManager.segments(), overlordClient, killUnusedConfig));
     }
-    if (config.isKillPendingSegmentsEnabled()) {
+    if (config.getKillConfigs().pendingSegments().isCleanupEnabled()) {
       duties.add(new KillStalePendingSegments(overlordClient));
     }
 
@@ -565,12 +570,17 @@ List<CoordinatorDuty> makeIndexingServiceDuties()
 
   private List<CoordinatorDuty> makeMetadataStoreManagementDuties()
   {
+    final CoordinatorKillConfigs killConfigs = config.getKillConfigs();
     return Arrays.asList(
-        new KillSupervisors(config, metadataManager.supervisors()),
-        new KillAuditLog(config, metadataManager.audit()),
-        new KillRules(config, metadataManager.rules()),
-        new KillDatasourceMetadata(config, metadataManager.indexer(), metadataManager.supervisors()),
-        new KillCompactionConfig(config, metadataManager.segments(), metadataManager.configs())
+        new KillSupervisors(killConfigs.supervisors(), metadataManager.supervisors()),
+        new KillAuditLog(killConfigs.audit(), metadataManager.audit()),
+        new KillRules(killConfigs.rules(), metadataManager.rules()),
+        new KillDatasourceMetadata(
+            killConfigs.datasource(),
+            metadataManager.indexer(),
+            metadataManager.supervisors()
+        ),
+        new KillCompactionConfig(killConfigs.compaction(), metadataManager.segments(), metadataManager.configs())
     );
   }
 

server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorKillConfigs.java

@@ -0,0 +1,146 @@
+/*
+ * 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.server.coordinator.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.joda.time.Duration;
+
+public class CoordinatorKillConfigs
+{
+  public static CoordinatorKillConfigs STANDARD
+      = new CoordinatorKillConfigs(null, null, null, null, null, null, null, null, null, null, null, null);
+
+  @JsonProperty("supervisor")
+  private final MetadataCleanupConfig supervisors;
+
+  @JsonProperty("audit")
+  private final MetadataCleanupConfig audit;
+
+  @JsonProperty("datasource")
+  private final MetadataCleanupConfig datasource;
+
+  @JsonProperty("rule")
+  private final MetadataCleanupConfig rules;
+
+  @JsonProperty("compaction")
+  private final MetadataCleanupConfig compaction;
+
+  @JsonProperty("pendingSegments")
+  private final MetadataCleanupConfig pendingSegments;
+
+  // Raw configs for killing unused segments
+  // These have been added as fields because KillUnusedSegmentsConfig is initialized lazily
+  @JsonProperty("on")
+  private final Boolean killUnusedEnabled;
+
+  @JsonProperty("period")
+  private final Duration killUnusedPeriod;
+
+  @JsonProperty("durationToRetain")
+  private final Duration killUnusedDurationToRetain;
+
+  @JsonProperty("ignoreDurationToRetain")
+  private final Boolean killUnusedIgnoreDurationToRetain;
+
+  @JsonProperty("bufferPeriod")
+  private final Duration killUnusedBufferPeriod;
+
+  @JsonProperty("maxSegments")
+  private final Integer killUnusedMaxSegments;
+
+  @JsonCreator
+  public CoordinatorKillConfigs(
+      @JsonProperty("pendingSegments") MetadataCleanupConfig pendingSegments,
+      @JsonProperty("supervisor") MetadataCleanupConfig supervisors,
+      @JsonProperty("audit") MetadataCleanupConfig audit,
+      @JsonProperty("datasource") MetadataCleanupConfig datasource,
+      @JsonProperty("rule") MetadataCleanupConfig rules,
+      @JsonProperty("compaction") MetadataCleanupConfig compaction,
+      // Configs for cleanup of unused segments
+      @JsonProperty("on") Boolean killUnusedEnabled,
+      @JsonProperty("period") Duration killUnusedPeriod,
+      @JsonProperty("durationToRetain") Duration killUnusedDurationToRetain,
+      @JsonProperty("ignoreDurationToRetain") Boolean killUnusedIgnoreDurationToRetain,
+      @JsonProperty("bufferPeriod") Duration killUnusedBufferPeriod,
+      @JsonProperty("maxSegments") Integer killUnusedMaxSegments
+  )
+  {
+    this.pendingSegments = Configs.valueOrDefault(pendingSegments, MetadataCleanupConfig.STANDARD);
+    this.supervisors = Configs.valueOrDefault(supervisors, MetadataCleanupConfig.STANDARD);
+    this.audit = Configs.valueOrDefault(audit, MetadataCleanupConfig.STANDARD);
+    this.datasource = Configs.valueOrDefault(datasource, MetadataCleanupConfig.STANDARD);
+    this.rules = Configs.valueOrDefault(rules, MetadataCleanupConfig.STANDARD);
+    this.compaction = Configs.valueOrDefault(compaction, MetadataCleanupConfig.STANDARD);
+
+    this.killUnusedEnabled = killUnusedEnabled;
+    this.killUnusedPeriod = killUnusedPeriod;
+    this.killUnusedDurationToRetain = killUnusedDurationToRetain;
+    this.killUnusedBufferPeriod = killUnusedBufferPeriod;
+    this.killUnusedIgnoreDurationToRetain = killUnusedIgnoreDurationToRetain;
+    this.killUnusedMaxSegments = killUnusedMaxSegments;
+  }
+
+  public MetadataCleanupConfig audit()
+  {
+    return audit;
+  }
+
+  public MetadataCleanupConfig datasource()
+  {
+    return datasource;
+  }
+
+  public MetadataCleanupConfig rules()
+  {
+    return rules;
+  }
+
+  public MetadataCleanupConfig compaction()
+  {
+    return compaction;
+  }
+
+  public MetadataCleanupConfig pendingSegments()
+  {
+    return pendingSegments;
+  }
+
+  public MetadataCleanupConfig supervisors()
+  {
+    return supervisors;
+  }
+
+  /**
+   * Creates a KillUnusedSegmentsConfig.
+   */
+  public KillUnusedSegmentsConfig unusedSegments(Duration indexingPeriod)
+  {
+    return new KillUnusedSegmentsConfig(
+        killUnusedEnabled,
+        Configs.valueOrDefault(killUnusedPeriod, indexingPeriod),
+        killUnusedDurationToRetain,
+        killUnusedIgnoreDurationToRetain,
+        killUnusedBufferPeriod,
+        killUnusedMaxSegments
+    );
+  }
+}