Change a logic according to a new decision

Author: Fly-Style

indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java

@@ -176,7 +176,7 @@ public boolean createOrUpdateAndStartSupervisor(SupervisorSpec spec)
       final boolean shouldUpdateSpec = shouldUpdateSupervisor(spec);
       SupervisorSpec existingSpec = possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
       if (existingSpec != null) {
-        spec.mergeSpecConfigs(existingSpec);
+        spec.merge(existingSpec);
       }
       createAndStartSupervisorInternal(spec, shouldUpdateSpec);
       return shouldUpdateSpec;

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -611,7 +611,8 @@ private void changeTaskCountInAutoScalerConfig(int desiredActiveTaskCount)
       log.warn("autoScalerConfig is null but scale action is submitted, how can it be ?");
       return;
     }
-    autoScalerConfig.setTaskCountStart(desiredActiveTaskCount);
+    ioConfig.setTaskCount(desiredActiveTaskCount);
+    // TODO: autoScalerConfig.setTaskCountStart(desiredActiveTaskCount);
     try {
       Optional<SupervisorManager> supervisorManager = taskMaster.getSupervisorManager();
       if (supervisorManager.isPresent()) {
@@ -1179,7 +1180,7 @@ public void stop(boolean stopGracefully)
       catch (Exception e) {
         stateManager.recordThrowableEvent(e);
         log.makeAlert(e, "Exception stopping [%s]", supervisorId)
-            .emit();
+           .emit();
       }
     }
   }
@@ -1357,7 +1358,7 @@ public void tryInit()
         }
         initRetryCounter++;
         log.makeAlert(e, "Exception starting SeekableStreamSupervisor[%s]", supervisorId)
-            .emit();
+           .emit();
 
         throw new RuntimeException(e);
       }

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java

@@ -263,29 +263,30 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
     }
   }
 
-  /**
-   * Writes the <code>taskCountStart</code> value from old config, if not specificed in new config.
-   *
-   * @param existingSpec the existing supervisor specification to merge configuration values from
-   */
   @Override
-  public void mergeSpecConfigs(@NotNull SupervisorSpec existingSpec)
+  public void merge(@NotNull SupervisorSpec existingSpec)
   {
     AutoScalerConfig thisAutoScalerConfig = this.getIoConfig().getAutoScalerConfig();
     // Either if autoscaler is absent or taskCountStart is specified - just return.
     if (thisAutoScalerConfig == null || thisAutoScalerConfig.getTaskCountStart() != null) {
       return;
     }
 
-    // TODO[sasha]: use switch expression with pattern matching when we move to Java 21 as minimum requirement.
+    // Use a switch expression with pattern matching when we move to Java 21 as a minimum requirement.
     if (existingSpec instanceof SeekableStreamSupervisorSpec) {
-      // Note: for some reason, sources are available only for bytecode version 11.
-      //noinspection PatternVariableCanBeUsed
-      var spec = (SeekableStreamSupervisorSpec) existingSpec;
+      SeekableStreamSupervisorSpec spec = (SeekableStreamSupervisorSpec) existingSpec;
       AutoScalerConfig autoScalerConfig = spec.getIoConfig().getAutoScalerConfig();
-      if (autoScalerConfig != null && autoScalerConfig.getTaskCountStart() != null) {
-        thisAutoScalerConfig.setTaskCountStart(autoScalerConfig.getTaskCountStart());
+      if (autoScalerConfig == null) {
+        return;
+      }
+      // provided `taskCountStart` > provided `taskCount` > existing `taskCount` > provided `taskCountMin`.
+      int taskCount = thisAutoScalerConfig.getTaskCountMin();
+      if (this.getIoConfig().getTaskCount() != null) {
+        taskCount = this.getIoConfig().getTaskCount();
+      } else if (spec.getIoConfig().getTaskCount() != null) {
+        taskCount = spec.getIoConfig().getTaskCount();
       }
+      this.getIoConfig().setTaskCount(taskCount);
     }
   }
 

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/AutoScalerConfig.java

@@ -40,7 +40,6 @@ public interface AutoScalerConfig
   int getTaskCountMax();
   int getTaskCountMin();
   Integer getTaskCountStart();
-  void setTaskCountStart(int taskCountStart);
   Double getStopTaskCountRatio();
   SupervisorTaskAutoScaler createAutoScaler(Supervisor supervisor, SupervisorSpec spec, ServiceEmitter emitter);
 }

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java

@@ -69,7 +69,8 @@ public LagBasedAutoScaler(
         .getLagCollectionIntervalMillis()) + 1;
     this.lagMetricsQueue = new CircularFifoQueue<>(slots);
     this.allocationExec = Execs.scheduledSingleThreaded(StringUtils.encodeForFormat(supervisorId) + "-Allocation-%d");
-    this.lagComputationExec = Execs.scheduledSingleThreaded(StringUtils.encodeForFormat(supervisorId) + "-Computation-%d");
+    this.lagComputationExec = Execs.scheduledSingleThreaded(StringUtils.encodeForFormat(supervisorId)
+                                                            + "-Computation-%d");
     this.spec = spec;
     this.supervisor = supervisor;
     this.emitter = emitter;
@@ -124,8 +125,10 @@ public void start()
     );
     log.info(
         "LagBasedAutoScaler will collect lag every [%d] millis and will keep up to [%d] data points for the last [%d] millis for dataSource [%s]",
-        lagBasedAutoScalerConfig.getLagCollectionIntervalMillis(), lagMetricsQueue.maxSize(),
-        lagBasedAutoScalerConfig.getLagCollectionRangeMillis(), dataSource
+        lagBasedAutoScalerConfig.getLagCollectionIntervalMillis(),
+        lagMetricsQueue.maxSize(),
+        lagBasedAutoScalerConfig.getLagCollectionRangeMillis(),
+        dataSource
     );
   }
 
@@ -190,27 +193,27 @@ private Runnable computeAndCollectLag()
     };
   }
 
-    /**
-     * This method determines whether to do scale actions based on collected lag points.
-     * The current algorithm of scale is straightforward:
-     * <ul>
-     * <li>First, compute the proportion of lag points higher/lower than <code>scaleOutThreshold</code>/<code>scaleInThreshold</code>,
-     * getting <code>scaleOutThreshold</code>/<code>scaleInThreshold</code>.
-     * <li>Secondly, compare <code>scaleOutThreshold</code>/<code>scaleInThreshold</code> with
-     * <code>triggerScaleOutFractionThreshold</code>/<code>triggerScaleInFractionThreshold</code>.
-     * <ul><li>P.S. Scale out action has a higher priority than scale in action.</ul>
-     * <li>Finally, if <code>scaleOutThreshold</code>/<code>scaleInThreshold</code> is higher than
-     * <code>triggerScaleOutFractionThreshold</code>/<code>triggerScaleInFractionThreshold</code>, scale out/in action would be triggered.
-     * </ul>
-     *
-     * @param lags the lag metrics of Stream (Kafka/Kinesis)
-     * @return Integer, target number of tasksCount. -1 means skip scale action.
-     */
+  /**
+   * This method determines whether to do scale actions based on collected lag points.
+   * The current algorithm of scale is straightforward:
+   * <ul>
+   * <li>First, compute the proportion of lag points higher/lower than {@code scaleOutThreshold/scaleInThreshold},
+   * getting {@code scaleInThreshold/scaleOutThreshold},.
+   * <li>Secondly, compare {@code scaleInThreshold/scaleOutThreshold} with
+   * {@code triggerScaleOutFractionThreshold/triggerScaleInFractionThreshold}.
+   * <ul><li>P.S. Scale out action has a higher priority than scale in action.</ul>
+   * <li>Finally, if {@code scaleOutThreshold/scaleInThreshold}, is higher than
+   * {@code triggerScaleOutFractionThreshold/triggerScaleInFractionThreshold}, scale out/in action would be triggered.
+   * </ul>
+   *
+   * @param lags the lag metrics of Stream (Kafka/Kinesis)
+   * @return Integer, target number of tasksCount. -1 means skip scale action.
+   */
   private int computeDesiredTaskCount(List<Long> lags)
   {
-    // if supervisor is not suspended, ensure required tasks are running
+    // if the supervisor is not suspended, ensure required tasks are running
     // if suspended, ensure tasks have been requested to gracefully stop
-    log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+    log.debug("Computing the desired task count for [%s], based on following lags : [%s]", dataSource, lags);
     int beyond = 0;
     int within = 0;
     int metricsCount = lags.size();
@@ -264,7 +267,8 @@ private int computeDesiredTaskCount(List<Long> lags)
       int taskCount = currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
       int actualTaskCountMin = Math.min(lagBasedAutoScalerConfig.getTaskCountMin(), partitionCount);
       if (currentActiveTaskCount == actualTaskCountMin) {
-        log.debug("CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
+        log.debug(
+            "CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
             dataSource
         );
         emitter.emit(metricBuilder

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerConfig.java

@@ -104,7 +104,7 @@ public LagBasedAutoScalerConfig(
     this.scaleInStep = scaleInStep != null ? scaleInStep : 1;
     this.scaleOutStep = scaleOutStep != null ? scaleOutStep : 2;
     this.minTriggerScaleActionFrequencyMillis = minTriggerScaleActionFrequencyMillis
-        != null ? minTriggerScaleActionFrequencyMillis : 600000;
+                                                != null ? minTriggerScaleActionFrequencyMillis : 600000;
 
     Preconditions.checkArgument(
         stopTaskCountRatio == null || (stopTaskCountRatio > 0.0 && stopTaskCountRatio <= 1.0),
@@ -183,12 +183,6 @@ public Integer getTaskCountStart()
     return taskCountStart;
   }
 
-  @Override
-  public void setTaskCountStart(int taskCountStart)
-  {
-    this.taskCountStart = taskCountStart;
-  }
-
   @Override
   public SupervisorTaskAutoScaler createAutoScaler(Supervisor supervisor, SupervisorSpec spec, ServiceEmitter emitter)
   {

indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamSupervisorSpecTest.java

@@ -1659,7 +1659,7 @@ public void testMergeSpecConfigs()
     Assert.assertNull(newSpec.getIoConfig().getAutoScalerConfig().getTaskCountStart());
 
     // When
-    newSpec.mergeSpecConfigs(existingSpec);
+    newSpec.merge(existingSpec);
 
     // Then - taskCountStart should be copied from existing spec
     Assert.assertEquals(Integer.valueOf(5), newSpec.getIoConfig().getAutoScalerConfig().getTaskCountStart());