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

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

@@ -56,6 +56,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;
@@ -189,7 +192,6 @@ public DruidCoordinator(
       ServiceAnnouncer serviceAnnouncer,
       @Self DruidNode self,
       CoordinatorCustomDutyGroups customDutyGroups,
-      BalancerStrategyFactory balancerStrategyFactory,
       LookupCoordinatorManager lookupCoordinatorManager,
       @Coordinator DruidLeaderSelector coordLeaderSelector,
       CompactionSegmentSearchPolicy compactionSegmentSearchPolicy,
@@ -209,7 +211,7 @@ public DruidCoordinator(
 
     this.executorFactory = scheduledExecutorFactory;
 
-    this.balancerStrategyFactory = balancerStrategyFactory;
+    this.balancerStrategyFactory = config.getBalancerStrategyFactory();
     this.lookupCoordinatorManager = lookupCoordinatorManager;
     this.coordLeaderSelector = coordLeaderSelector;
     this.compactSegments = initializeCompactSegmentsDuty(compactionSegmentSearchPolicy);
@@ -562,10 +564,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));
     }
 
@@ -582,16 +587,23 @@ List<CoordinatorDuty> makeIndexingServiceDuties()
 
   private List<CoordinatorDuty> makeMetadataStoreManagementDuties()
   {
-    List<CoordinatorDuty> duties = new ArrayList<>();
-
-    duties.add(new KillSupervisors(config, metadataManager.supervisors()));
-    duties.add(new KillAuditLog(config, metadataManager.audit()));
-    duties.add(new KillRules(config, metadataManager.rules()));
-    duties.add(new KillDatasourceMetadata(config, metadataManager.indexer(), metadataManager.supervisors()));
-    duties.add(new KillCompactionConfig(config, metadataManager.segments(), metadataManager.configs()));
-
+    final CoordinatorKillConfigs killConfigs = config.getKillConfigs();
+    final List<CoordinatorDuty> duties = new ArrayList<>();
+    duties.add(new KillSupervisors(killConfigs.supervisors(), metadataManager.supervisors()));
+    duties.add(new KillAuditLog(killConfigs.auditLogs(), metadataManager.audit()));
+    duties.add(new KillRules(killConfigs.rules(), metadataManager.rules()));
+    duties.add(
+        new KillDatasourceMetadata(
+            killConfigs.datasources(),
+            metadataManager.indexer(),
+            metadataManager.supervisors()
+        )
+    );
+    duties.add(
+        new KillCompactionConfig(killConfigs.compactionConfigs(), metadataManager.segments(), metadataManager.configs())
+    );
     if (centralizedDatasourceSchemaConfig.isEnabled()) {
-      duties.add(new KillUnreferencedSegmentSchemaDuty(config, metadataManager.schemas()));
+      duties.add(new KillUnreferencedSegmentSchemaDuty(killConfigs.segmentSchemas(), metadataManager.schemas()));
     }
     return duties;
   }