Switch to minScale.*Delay configs and deprecate minTriggerScaleActionFrequencyMillis

Author: jtuglu1

docs/ingestion/kafka-ingestion.md

@@ -235,7 +235,8 @@ The following example shows a supervisor spec with idle configuration enabled:
         "enableTaskAutoScaler": true,
         "taskCountMax": 6,
         "taskCountMin": 2,
-        "minTriggerScaleActionFrequencyMillis": 600000,
+        "minScaleUpDelay": "PT10M",
+        "minScaleDownDelay": "PT10M",
         "autoScalerStrategy": "lagBased",
         "lagCollectionIntervalMillis": 30000,
         "lagCollectionRangeMillis": 600000,

docs/ingestion/supervisor.md

@@ -79,9 +79,9 @@ The following table outlines the configuration properties for `autoScalerConfig`
 |`taskCountMax`|The maximum number of ingestion tasks. Must be greater than or equal to `taskCountMin`. If `taskCountMax` is greater than the number of Kafka partitions or Kinesis shards, Druid sets the maximum number of reading tasks to the number of Kafka partitions or Kinesis shards and ignores `taskCountMax`.|Yes||
 |`taskCountMin`|The minimum number of ingestion tasks. When you enable the autoscaler, Druid computes the initial number of tasks to launch by checking the configs in the following order: `taskCountStart`, then `taskCount` (in `ioConfig`), then `taskCountMin`.|Yes||
 |`taskCountStart`|Optional config to specify the number of ingestion tasks to start with. When you enable the autoscaler, Druid computes the initial number of tasks to launch by checking the configs in the following order: `taskCountStart`, then `taskCount` (in `ioConfig`), then `taskCountMin`.|No|`taskCount` or `taskCountMin`|
-|`minTriggerScaleActionFrequencyMillis`|The minimum time interval between any two scale actions. Used as the default fallback when `minTriggerScaleUpActionFrequencyMillis` or `minTriggerScaleDownActionFrequencyMillis` are not set.| No|600000|
-|`minTriggerScaleUpActionFrequencyMillis`|The minimum time interval between two scale-up actions. Falls back to `minTriggerScaleActionFrequencyMillis` if not set.| No||
-|`minTriggerScaleDownActionFrequencyMillis`|The minimum time interval between two scale-down actions. Falls back to `minTriggerScaleActionFrequencyMillis` if not set.| No||
+|`minScaleUpDelay`|Minimum cooldown duration between scale-up actions, specified as an ISO-8601 duration string. Falls back to `minTriggerScaleActionFrequencyMillis` if not set.|No||
+|`minScaleDownDelay`|Minimum cooldown duration between scale-down actions, specified as an ISO-8601 duration string. Falls back to `minTriggerScaleActionFrequencyMillis` if not set.|No||
+|`minTriggerScaleActionFrequencyMillis`|**Deprecated.** Use `minScaleUpDelay` and `minScaleDownDelay` instead. Minimum time interval in milliseconds between scale actions, used as the fallback when the Duration-based fields are not set.|No|600000|
 |`autoScalerStrategy`|The algorithm of autoscaler. Druid only supports the `lagBased` strategy. See [Autoscaler strategy](#autoscaler-strategy) for more information.|No|`lagBased`|
 |`stopTaskCountRatio`|A variable version of `ioConfig.stopTaskCount` with a valid range of (0.0, 1.0]. Allows the maximum number of stoppable tasks in steady state to be proportional to the number of tasks currently running.|No||
 
@@ -163,7 +163,8 @@ The following example shows a supervisor spec with `lagBased` autoscaler:
       "enableTaskAutoScaler": true,
       "taskCountMax": 6,
       "taskCountMin": 2,
-      "minTriggerScaleActionFrequencyMillis": 600000,
+      "minScaleUpDelay": "PT10M",
+      "minScaleDownDelay": "PT10M",
       "autoScalerStrategy": "lagBased",
       "lagCollectionIntervalMillis": 30000,
       "lagCollectionRangeMillis": 600000,
@@ -212,10 +213,11 @@ The following table outlines the configuration properties related to the `costBa
 |`idleWeight`|The weight of extracted poll idle value in cost function. | No | 0.75 |
 |`useTaskCountBoundaries`|Enables the bounded partitions-per-task window when selecting task counts.|No| `false` |
 |`highLagThreshold`|Average partition lag threshold that triggers burst scale-up when set to a value greater than `0`. Set to a negative value to disable burst scale-up.|No|-1|
-|`minScaleDownDelay`|Minimum duration between successful scale actions, specified as an ISO-8601 duration string.|No|`PT30M`|
+|`minScaleUpDelay`|Minimum cooldown duration after a scale-up action before the next scale-up is allowed, specified as an ISO-8601 duration string. Falls back to `minScaleDownDelay` if not set.|No||
+|`minScaleDownDelay`|Minimum cooldown duration after a scale-down action before the next scale-down is allowed, specified as an ISO-8601 duration string.|No|`PT30M`|
 |`scaleDownDuringTaskRolloverOnly`|Indicates whether task scaling down is limited to periods during task rollovers only.|No|`false`|
 
-The following example shows a supervisor spec with `lagBased` autoscaler:
+The following example shows a supervisor spec with `costBased` autoscaler:
 
 <details>
   <summary>Click to view the example</summary>
@@ -229,9 +231,10 @@ The following example shows a supervisor spec with `lagBased` autoscaler:
       "autoScalerStrategy": "costBased",
       "taskCountMin": 1,
       "taskCountMax": 10,
-      "minTriggerScaleActionFrequencyMillis": 600000,
+      "minScaleUpDelay": "PT10M",
+      "minScaleDownDelay": "PT30M",
       "lagWeight": 0.1,
-      "idleWeight": 0.9,
+      "idleWeight": 0.9
     }
   }
 }

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfigTest.java

@@ -315,6 +315,8 @@ public void testAutoScalingConfigSerde() throws JsonProcessingException
     autoScalerConfig.put("scaleInStep", 1);
     autoScalerConfig.put("scaleOutStep", 2);
     autoScalerConfig.put("minTriggerScaleActionFrequencyMillis", 1200000);
+    autoScalerConfig.put("minScaleUpDelay", "PT20M");
+    autoScalerConfig.put("minScaleDownDelay", "PT20M");
 
     final Map<String, Object> consumerProperties = KafkaConsumerConfigs.getConsumerProperties();
     consumerProperties.put("bootstrap.servers", "localhost:8082");

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

@@ -306,7 +306,8 @@ public SeekableStreamIndexTaskClient<KafkaTopicPartition, Long> build(
     autoScalerConfig.put("taskCountMin", 1);
     autoScalerConfig.put("scaleInStep", 1);
     autoScalerConfig.put("scaleOutStep", 2);
-    autoScalerConfig.put("minTriggerScaleActionFrequencyMillis", 1200000);
+    autoScalerConfig.put("minScaleUpDelay", "PT20M");
+    autoScalerConfig.put("minScaleDownDelay", "PT20M");
 
     final Map<String, Object> consumerProperties = KafkaConsumerConfigs.getConsumerProperties();
     consumerProperties.put("myCustomKey", "myCustomValue");

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java

@@ -310,7 +310,8 @@ public void testNoInitialStateWithAutoScaleOut() throws Exception
     autoScalerConfigMap.put("taskCountMin", 1);
     autoScalerConfigMap.put("scaleInStep", 1);
     autoScalerConfigMap.put("scaleOutStep", 2);
-    autoScalerConfigMap.put("minTriggerScaleActionFrequencyMillis", 1200000);
+    autoScalerConfigMap.put("minScaleUpDelay", "PT20M");
+    autoScalerConfigMap.put("minScaleDownDelay", "PT20M");
 
     AutoScalerConfig autoScalerConfig = OBJECT_MAPPER.convertValue(autoScalerConfigMap, AutoScalerConfig.class);
     supervisor = getTestableSupervisor(
@@ -384,7 +385,8 @@ public void testNoInitialStateWithAutoScaleIn() throws Exception
     autoScalerConfigMap.put("taskCountMin", 1);
     autoScalerConfigMap.put("scaleInStep", 1);
     autoScalerConfigMap.put("scaleOutStep", 2);
-    autoScalerConfigMap.put("minTriggerScaleActionFrequencyMillis", 1200000);
+    autoScalerConfigMap.put("minScaleUpDelay", "PT20M");
+    autoScalerConfigMap.put("minScaleDownDelay", "PT20M");
 
     AutoScalerConfig autoScalerConfig = OBJECT_MAPPER.convertValue(autoScalerConfigMap, AutoScalerConfig.class);
     supervisor = getTestableSupervisor(

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

@@ -552,31 +552,23 @@ public void handle()
           // configure a longer cooldown for scale-down than for scale-up. Both directions are measured against the same
           // last-scale timestamp so that a rapid up/down oscillation is still subject to the appropriate cooldown,
           // regardless of which direction triggered last.
-          final long effectiveFreq;
           final ScaleDirection scaleDirection;
+          final long cooldownMillis;
 
           if (desiredTaskCount > currentTaskCount) {
-            // Scale up: use the scale-up specific frequency, falling back to base.
-            Long specificFreq = autoScalerConfig.getMinTriggerScaleUpActionFrequencyMillis();
-            effectiveFreq = specificFreq != null
-                            ? specificFreq
-                            : autoScalerConfig.getMinTriggerScaleActionFrequencyMillis();
             scaleDirection = ScaleDirection.SCALE_UP;
+            cooldownMillis = autoScalerConfig.getMinScaleUpDelay().getMillis();
           } else {
-            // Scale down: use the scale-down specific frequency, falling back to base.
-            Long specificFreq = autoScalerConfig.getMinTriggerScaleDownActionFrequencyMillis();
-            effectiveFreq = specificFreq != null
-                            ? specificFreq
-                            : autoScalerConfig.getMinTriggerScaleActionFrequencyMillis();
             scaleDirection = ScaleDirection.SCALE_DOWN;
+            cooldownMillis = autoScalerConfig.getMinScaleDownDelay().getMillis();
           }
 
-          if (nowTime - dynamicTriggerLastScaleRunTime < effectiveFreq) {
+          if (nowTime - dynamicTriggerLastScaleRunTime < cooldownMillis) {
             log.info(
-                "DynamicAllocationTasksNotice submitted again in [%d]ms, effective [%s] throttle is [%d]ms for supervisor[%s] for dataSource[%s], skipping it! desired task count is [%d], current task count is [%d]",
+                "DynamicAllocationTasksNotice submitted again in [%d]ms, [%s] cooldown is [%d]ms for supervisor[%s] for dataSource[%s], skipping it! desired task count is [%d], current task count is [%d]",
                 nowTime - dynamicTriggerLastScaleRunTime,
                 scaleDirection,
-                effectiveFreq,
+                cooldownMillis,
                 supervisorId,
                 dataSource,
                 desiredTaskCount,
@@ -585,7 +577,7 @@ public void handle()
 
             emitter.emit(event.setDimension(
                                   AUTOSCALER_SKIP_REASON_DIMENSION,
-                                  "minTriggerScaleActionFrequencyMillis not elapsed yet"
+                                  "Scale cooldown not elapsed yet"
                               )
                               .setMetric(AUTOSCALER_REQUIRED_TASKS_METRIC, desiredTaskCount));
             return;
@@ -633,8 +625,8 @@ public String getType()
    *
    * @param desiredActiveTaskCount desired taskCount computed from AutoScaler
    * @return Boolean flag indicating if scale action was executed or not. If true, it will wait at least the configured
-   * minTriggerScale(Up|Down)ActionFrequencyMillis (falling back to minTriggerScaleActionFrequencyMillis) before the next same-direction 'changeTaskCount'.
-   * If false, it will do 'changeTaskCount' again after 'scaleActionPeriodMillis' millis.
+   * minScaleUpDelay or minScaleDownDelay (falling back to minTriggerScaleActionFrequencyMillis) before the next
+   * same-direction 'changeTaskCount'. If false, it will do 'changeTaskCount' again after 'scaleActionPeriodMillis' millis.
    * @throws InterruptedException
    * @throws ExecutionException
    */

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

@@ -27,8 +27,7 @@
 import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
 import org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
-
-import javax.annotation.Nullable;
+import org.joda.time.Duration;
 
 @UnstableApi
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "autoScalerStrategy", defaultImpl = LagBasedAutoScalerConfig.class)
@@ -39,19 +38,27 @@
 public interface AutoScalerConfig
 {
   boolean getEnableTaskAutoScaler();
+
+  /**
+   * @deprecated Use {@link #getMinScaleUpDelay()} and {@link #getMinScaleDownDelay()} instead.
+   *             This field is retained for backward compatibility and will be removed in a future version.
+   */
+  @Deprecated
   long getMinTriggerScaleActionFrequencyMillis();
 
-  @Nullable
-  default Long getMinTriggerScaleUpActionFrequencyMillis()
-  {
-    return null;
-  }
-
-  @Nullable
-  default Long getMinTriggerScaleDownActionFrequencyMillis()
-  {
-    return null;
-  }
+  /**
+   * Minimum time that must elapse after any scale action before a scale-up is permitted.
+   * If not explicitly configured, implementations fall back to
+   * {@link #getMinTriggerScaleActionFrequencyMillis()} for backward compatibility.
+   */
+  Duration getMinScaleUpDelay();
+
+  /**
+   * Minimum time that must elapse after any scale action before a scale-down is permitted.
+   * If not explicitly configured, implementations fall back to
+   * {@link #getMinTriggerScaleActionFrequencyMillis()} for backward compatibility.
+   */
+  Duration getMinScaleDownDelay();
 
   int getTaskCountMax();
   int getTaskCountMin();

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

@@ -179,12 +179,12 @@ public int computeTaskCountForScaleAction()
 
     // Perform scale-up actions; scale-down actions only if configured.
     final int taskCount;
-    if (isScaleActionAllowed() && optimalTaskCount > currentTaskCount) {
+    if (isScaleActionAllowed(true) && optimalTaskCount > currentTaskCount) {
       taskCount = optimalTaskCount;
       lastScaleActionTimeMillis = DateTimes.nowUtc().getMillis();
       log.info("Updating taskCount for supervisor[%s] from [%d] to [%d] (scale up).", supervisorId, currentTaskCount, taskCount);
     } else if (!config.isScaleDownOnTaskRolloverOnly()
-               && isScaleActionAllowed()
+               && isScaleActionAllowed(false)
                && optimalTaskCount < currentTaskCount
                && optimalTaskCount > 0) {
       taskCount = optimalTaskCount;
@@ -562,16 +562,18 @@ private Either<String, Boolean> validateMetricsForScaling(CostMetrics metrics)
   }
 
   /**
-   * Determines if a scale action is currently allowed based on the elapsed time
-   * since the last scale action and the configured minimum scale-down delay.
+   * Determines if a scale action in the given direction is currently allowed based on the elapsed time
+   * since the last scale action and the configured delay for that direction.
    */
-  private boolean isScaleActionAllowed()
+  private boolean isScaleActionAllowed(boolean isScaleUp)
   {
     if (lastScaleActionTimeMillis < 0) {
       return true;
     }
 
-    final long barrierMillis = config.getMinScaleDownDelay().getMillis();
+    final long barrierMillis = isScaleUp
+                               ? config.getMinScaleUpDelay().getMillis()
+                               : config.getMinScaleDownDelay().getMillis();
     if (barrierMillis <= 0) {
       return true;
     }