Reduce coordinator logs when operating normally (#14926)

Changes:
- Reduce log level of some coordinator stats, which only denote normal coordinator operation.
These stats are still emitted and can be logged by setting debugDimensions in the coordinator
dynamic config.
- Initialize SegmentLoadingConfig only for historical management duties. This config is not
needed in other duties and initializing it creates logs which are misleading.

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

@@ -19,7 +19,6 @@
 
 package org.apache.druid.server.coordinator;
 
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.client.DataSourcesSnapshot;
@@ -256,7 +255,6 @@ private Builder(
     public DruidCoordinatorRuntimeParams build()
     {
       initStatsIfRequired();
-      initSegmentLoadingConfigIfRequired();
       initSegmentAssignerIfRequired();
 
       return new DruidCoordinatorRuntimeParams(
@@ -282,15 +280,10 @@ private void initStatsIfRequired()
       stats = stats == null ? new CoordinatorRunStats(debugDimensions) : stats;
     }
 
-    private void initSegmentLoadingConfigIfRequired()
-    {
-      if (segmentLoadingConfig == null
-          && coordinatorDynamicConfig != null
-          && usedSegments != null) {
-        segmentLoadingConfig = SegmentLoadingConfig.create(coordinatorDynamicConfig, usedSegments.size());
-      }
-    }
-
+    /**
+     * Initializes {@link StrategicSegmentAssigner} used by historical management
+     * duties for segment load/drop/move.
+     */
     private void initSegmentAssignerIfRequired()
     {
       if (segmentAssigner != null || loadQueueManager == null) {
@@ -299,8 +292,13 @@ private void initSegmentAssignerIfRequired()
 
       Preconditions.checkNotNull(druidCluster);
       Preconditions.checkNotNull(balancerStrategy);
-      Preconditions.checkNotNull(segmentLoadingConfig);
+      Preconditions.checkNotNull(usedSegments);
       Preconditions.checkNotNull(stats);
+
+      if (segmentLoadingConfig == null) {
+        segmentLoadingConfig = SegmentLoadingConfig.create(coordinatorDynamicConfig, usedSegments.size());
+      }
+
       segmentAssigner = new StrategicSegmentAssigner(
           loadQueueManager,
           druidCluster,
@@ -339,16 +337,12 @@ public Builder withDataSourcesSnapshot(DataSourcesSnapshot snapshot)
       return this;
     }
 
-    /** This method must be used in test code only. */
-    @VisibleForTesting
-    public Builder withUsedSegmentsInTest(DataSegment... usedSegments)
+    public Builder withUsedSegments(DataSegment... usedSegments)
     {
-      return withUsedSegmentsInTest(Arrays.asList(usedSegments));
+      return withUsedSegments(Arrays.asList(usedSegments));
     }
 
-    /** This method must be used in test code only. */
-    @VisibleForTesting
-    public Builder withUsedSegmentsInTest(Collection<DataSegment> usedSegments)
+    public Builder withUsedSegments(Collection<DataSegment> usedSegments)
     {
       this.usedSegments = createUsedSegmentsSet(usedSegments);
       this.dataSourcesSnapshot = DataSourcesSnapshot.fromUsedSegments(usedSegments, ImmutableMap.of());
@@ -361,6 +355,12 @@ public Builder withDynamicConfigs(CoordinatorDynamicConfig configs)
       return this;
     }
 
+    public Builder withSegmentLoadingConfig(SegmentLoadingConfig config)
+    {
+      this.segmentLoadingConfig = config;
+      return this;
+    }
+
     public Builder withCompactionConfig(CoordinatorCompactionConfig config)
     {
       this.coordinatorCompactionConfig = config;

server/src/main/java/org/apache/druid/server/coordinator/duty/PrepareBalancerAndLoadQueues.java

@@ -84,7 +84,8 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     taskMaster.resetPeonsForNewServers(currentServers);
 
     final CoordinatorDynamicConfig dynamicConfig = params.getCoordinatorDynamicConfig();
-    final SegmentLoadingConfig segmentLoadingConfig = params.getSegmentLoadingConfig();
+    final SegmentLoadingConfig segmentLoadingConfig
+        = SegmentLoadingConfig.create(dynamicConfig, params.getUsedSegments().size());
 
     final DruidCluster cluster = prepareCluster(dynamicConfig, segmentLoadingConfig, currentServers);
     cancelLoadsOnDecommissioningServers(cluster);
@@ -103,6 +104,7 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     return params.buildFromExisting()
                  .withDruidCluster(cluster)
                  .withBalancerStrategy(balancerStrategy)
+                 .withSegmentLoadingConfig(segmentLoadingConfig)
                  .withSegmentAssignerUsing(loadQueueManager)
                  .build();
   }

server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java

@@ -28,9 +28,9 @@ public static class Segments
   {
     // Decisions taken in a run
     public static final CoordinatorStat ASSIGNED
-        = CoordinatorStat.toLogAndEmit("assigned", "segment/assigned/count", CoordinatorStat.Level.INFO);
+        = CoordinatorStat.toDebugAndEmit("assigned", "segment/assigned/count");
     public static final CoordinatorStat DROPPED
-        = CoordinatorStat.toLogAndEmit("dropped", "segment/dropped/count", CoordinatorStat.Level.INFO);
+        = CoordinatorStat.toDebugAndEmit("dropped", "segment/dropped/count");
     public static final CoordinatorStat DELETED
         = CoordinatorStat.toLogAndEmit("deleted", "segment/deleted/count", CoordinatorStat.Level.INFO);
     public static final CoordinatorStat MOVED

server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsProfiler.java

@@ -130,7 +130,7 @@ public void bigProfiler()
     DruidCoordinatorRuntimeParams params = DruidCoordinatorRuntimeParams
         .newBuilder(DateTimes.nowUtc())
         .withDruidCluster(druidCluster)
-        .withUsedSegmentsInTest(segments)
+        .withUsedSegments(segments)
         .withDynamicConfigs(
             CoordinatorDynamicConfig
                 .builder()
@@ -185,7 +185,7 @@ public void profileRun()
                 )
                 .build()
         )
-        .withUsedSegmentsInTest(segments)
+        .withUsedSegments(segments)
         .withDynamicConfigs(CoordinatorDynamicConfig.builder().withMaxSegmentsToMove(MAX_SEGMENTS_TO_MOVE).build())
         .withSegmentAssignerUsing(loadQueueManager)
         .build();

server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java

@@ -43,6 +43,7 @@
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
 import org.apache.druid.java.util.emitter.core.Event;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.metadata.MetadataRuleManager;
@@ -80,7 +81,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -165,7 +165,7 @@ public void setUp() throws Exception
     );
     loadQueuePeon.start();
     druidNode = new DruidNode("hey", "what", false, 1234, null, true, false);
-    scheduledExecutorFactory = (corePoolSize, nameFormat) -> Executors.newSingleThreadScheduledExecutor();
+    scheduledExecutorFactory = ScheduledExecutors::fixed;
     leaderAnnouncerLatch = new CountDownLatch(1);
     leaderUnannouncerLatch = new CountDownLatch(1);
     coordinator = new DruidCoordinator(

server/src/test/java/org/apache/druid/server/coordinator/duty/BalanceSegmentsTest.java

@@ -395,7 +395,7 @@ private DruidCoordinatorRuntimeParams.Builder defaultRuntimeParamsBuilder(
     return DruidCoordinatorRuntimeParams
         .newBuilder(DateTimes.nowUtc())
         .withDruidCluster(DruidCluster.builder().addTier("normal", servers).build())
-        .withUsedSegmentsInTest(allSegments)
+        .withUsedSegments(allSegments)
         .withBroadcastDatasources(broadcastDatasources)
         .withBalancerStrategy(balancerStrategy)
         .withSegmentAssignerUsing(loadQueueManager);

server/src/test/java/org/apache/druid/server/coordinator/duty/CollectSegmentAndServerStatsTest.java

@@ -48,7 +48,7 @@ public void testCollectedSegmentStats()
     DruidCoordinatorRuntimeParams runtimeParams =
         DruidCoordinatorRuntimeParams.newBuilder(DateTimes.nowUtc())
                                      .withDruidCluster(DruidCluster.EMPTY)
-                                     .withUsedSegmentsInTest()
+                                     .withUsedSegments()
                                      .withBalancerStrategy(new RandomBalancerStrategy())
                                      .withSegmentAssignerUsing(new SegmentLoadQueueManager(null, null))
                                      .build();

server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java

@@ -337,7 +337,7 @@ private DruidCoordinatorRuntimeParams.Builder createCoordinatorRuntimeParams(
     return DruidCoordinatorRuntimeParams
         .newBuilder(DateTimes.nowUtc().minusDays(1))
         .withDruidCluster(druidCluster)
-        .withUsedSegmentsInTest(dataSegments)
+        .withUsedSegments(dataSegments)
         .withDatabaseRuleManager(databaseRuleManager);
   }
 
@@ -830,7 +830,7 @@ public void testNoThrottleWhenSegmentNotLoadedInTier()
 
     stats = runDutyAndGetStats(
         createCoordinatorRuntimeParams(druidCluster)
-            .withUsedSegmentsInTest(overFlowSegment)
+            .withUsedSegments(overFlowSegment)
             .withBalancerStrategy(balancerStrategy)
             .withSegmentAssignerUsing(loadQueueManager)
             .build()
@@ -950,7 +950,7 @@ public void testDropReplicantThrottle()
                     .build();
 
     DruidCoordinatorRuntimeParams params = createCoordinatorRuntimeParams(druidCluster)
-        .withUsedSegmentsInTest(longerUsedSegments)
+        .withUsedSegments(longerUsedSegments)
         .withBalancerStrategy(new CostBalancerStrategy(balancerExecutor))
         .withSegmentAssignerUsing(loadQueueManager)
         .build();
@@ -1004,7 +1004,7 @@ public void testRulesRunOnNonOvershadowedSegmentsOnly()
     ).build();
 
     DruidCoordinatorRuntimeParams params = createCoordinatorRuntimeParams(druidCluster)
-        .withUsedSegmentsInTest(usedSegments)
+        .withUsedSegments(usedSegments)
         .withBalancerStrategy(new CostBalancerStrategy(balancerExecutor))
         .withDynamicConfigs(CoordinatorDynamicConfig.builder().withMaxSegmentsToMove(5).build())
         .withSegmentAssignerUsing(loadQueueManager)

server/src/test/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegmentsTest.java

@@ -259,7 +259,7 @@ public void test_unloadUnusedSegmentsFromAllServers()
                 .addRealtimes(new ServerHolder(indexerServer, indexerPeon, false))
                 .build()
         )
-        .withUsedSegmentsInTest(usedSegments)
+        .withUsedSegments(usedSegments)
         .withBroadcastDatasources(Collections.singleton(broadcastDatasource))
         .withDatabaseRuleManager(databaseRuleManager)
         .build();

server/src/test/java/org/apache/druid/server/coordinator/rules/BroadcastDistributionRuleTest.java

@@ -236,7 +236,7 @@ private DruidCoordinatorRuntimeParams makeParamsWithUsedSegments(
     return DruidCoordinatorRuntimeParams
         .newBuilder(DateTimes.nowUtc())
         .withDruidCluster(druidCluster)
-        .withUsedSegmentsInTest(usedSegments)
+        .withUsedSegments(usedSegments)
         .withBalancerStrategy(new RandomBalancerStrategy())
         .withSegmentAssignerUsing(new SegmentLoadQueueManager(null, null))
         .build();

server/src/test/java/org/apache/druid/server/coordinator/rules/LoadRuleTest.java

@@ -145,7 +145,7 @@ private DruidCoordinatorRuntimeParams makeCoordinatorRuntimeParams(
         .newBuilder(DateTimes.nowUtc())
         .withDruidCluster(druidCluster)
         .withBalancerStrategy(balancerStrategy)
-        .withUsedSegmentsInTest(usedSegments)
+        .withUsedSegments(usedSegments)
         .withDynamicConfigs(
             CoordinatorDynamicConfig.builder()
                                     .withSmartSegmentLoading(false)
@@ -335,7 +335,7 @@ public void testMaxLoadingQueueSize()
         .newBuilder(DateTimes.nowUtc())
         .withDruidCluster(druidCluster)
         .withBalancerStrategy(balancerStrategy)
-        .withUsedSegmentsInTest(dataSegment1, dataSegment2, dataSegment3)
+        .withUsedSegments(dataSegment1, dataSegment2, dataSegment3)
         .withDynamicConfigs(
             CoordinatorDynamicConfig.builder()
                                     .withSmartSegmentLoading(false)