You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by tu...@apache.org on 2025/09/08 03:59:21 UTC
(druid) branch master updated: Add stopTaskCountRatio field to LagBasedAutoscaler (#18480)
This is an automated email from the ASF dual-hosted git repository.
tuglu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new cb04e71d6a1 Add stopTaskCountRatio field to LagBasedAutoscaler (#18480)
cb04e71d6a1 is described below
commit cb04e71d6a1ec43b062c707e0037a859e5656bb6
Author: jtuglu1 <jt...@netflix.com>
AuthorDate: Sun Sep 7 20:59:15 2025 -0700
Add stopTaskCountRatio field to LagBasedAutoscaler (#18480)
- A variable version of `ioConfig.stopTaskCount` with a valid range of (0.0, 1.0].
- Allows the maximum number of stopped tasks to scale proportional to the task autoscaler.
---
docs/ingestion/supervisor.md | 1 +
.../SeekableStreamSupervisorIOConfig.java | 9 +-
.../supervisor/autoscaler/AutoScalerConfig.java | 1 +
.../autoscaler/LagBasedAutoScalerConfig.java | 81 ++++-
.../autoscaler/LagBasedAutoScalerConfigTest.java | 331 +++++++++++++++++++
.../SeekableStreamSupervisorIOConfigTest.java | 356 +++++++++++++++++++++
.../SeekableStreamSupervisorStateTest.java | 50 +++
7 files changed, 824 insertions(+), 5 deletions(-)
diff --git a/docs/ingestion/supervisor.md b/docs/ingestion/supervisor.md
index 50e8857b00d..882207cac16 100644
--- a/docs/ingestion/supervisor.md
+++ b/docs/ingestion/supervisor.md
@@ -80,6 +80,7 @@ The following table outlines the configuration properties for `autoScalerConfig`
|`taskCountStart`|Optional config to specify the number of ingestion tasks to start with. When you enable the autoscaler, Druid ignores the value of `taskCount` in `ioConfig` and, if specified, starts with the `taskCountStart` number of tasks. Otherwise, defaults to `taskCountMin`.|No|`taskCountMin`|
|`minTriggerScaleActionFrequencyMillis`|The minimum time interval between two scale actions.| 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||
##### Autoscaler strategy
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java
index a28f06d337e..8a7fcad1511 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java
@@ -53,6 +53,7 @@ public abstract class SeekableStreamSupervisorIOConfig
@Nullable private final Integer stopTaskCount;
private final LagAggregator lagAggregator;
+ private final boolean autoScalerEnabled;
public SeekableStreamSupervisorIOConfig(
String stream,
@@ -84,8 +85,9 @@ public abstract class SeekableStreamSupervisorIOConfig
this.lagAggregator = lagAggregator;
// Could be null
this.autoScalerConfig = autoScalerConfig;
- // if autoscaler is enable then taskcount will be ignored here. and init taskcount will be equal to taskCountMin
- if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
+ this.autoScalerEnabled = autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler();
+ // if autoscaler is enabled then taskCount will be ignored here and initial taskCount will equal to taskCountStart/taskCountMin
+ if (autoScalerEnabled) {
final Integer startTaskCount = autoScalerConfig.getTaskCountStart();
this.taskCount = startTaskCount != null ? startTaskCount : autoScalerConfig.getTaskCountMin();
} else {
@@ -231,6 +233,9 @@ public abstract class SeekableStreamSupervisorIOConfig
public int getMaxAllowedStops()
{
+ if (autoScalerEnabled && autoScalerConfig.getStopTaskCountRatio() != null) {
+ return (int) Math.max(1, Math.floor(taskCount * autoScalerConfig.getStopTaskCountRatio()));
+ }
return stopTaskCount == null ? taskCount : stopTaskCount;
}
}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/AutoScalerConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/AutoScalerConfig.java
index afd3d0dcfc1..b9eb741bad1 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/AutoScalerConfig.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/AutoScalerConfig.java
@@ -40,6 +40,7 @@ public interface AutoScalerConfig
int getTaskCountMax();
int getTaskCountMin();
Integer getTaskCountStart();
+ Double getStopTaskCountRatio();
SupervisorTaskAutoScaler createAutoScaler(Supervisor supervisor, SupervisorSpec spec, ServiceEmitter emitter);
}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerConfig.java
index 60c7ff0ea9e..b4a9b0e8891 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerConfig.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerConfig.java
@@ -22,6 +22,7 @@ package org.apache.druid.indexing.seekablestream.supervisor.autoscaler;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
import org.apache.druid.indexing.overlord.supervisor.Supervisor;
import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
import org.apache.druid.indexing.overlord.supervisor.autoscaler.AggregateFunction;
@@ -30,7 +31,9 @@ import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervi
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import javax.annotation.Nullable;
+import java.util.Objects;
+@JsonInclude(JsonInclude.Include.NON_NULL)
public class LagBasedAutoScalerConfig implements AutoScalerConfig
{
private final long lagCollectionIntervalMillis;
@@ -43,12 +46,13 @@ public class LagBasedAutoScalerConfig implements AutoScalerConfig
private final double triggerScaleInFractionThreshold;
private int taskCountMax;
private int taskCountMin;
- @Nullable private Integer taskCountStart;
+ private Integer taskCountStart;
private final int scaleInStep;
private final int scaleOutStep;
private final boolean enableTaskAutoScaler;
private final long minTriggerScaleActionFrequencyMillis;
private final AggregateFunction lagAggregate;
+ private final Double stopTaskCountRatio;
@JsonCreator
public LagBasedAutoScalerConfig(
@@ -67,7 +71,8 @@ public class LagBasedAutoScalerConfig implements AutoScalerConfig
@Nullable @JsonProperty("scaleOutStep") Integer scaleOutStep,
@Nullable @JsonProperty("enableTaskAutoScaler") Boolean enableTaskAutoScaler,
@Nullable @JsonProperty("minTriggerScaleActionFrequencyMillis") Long minTriggerScaleActionFrequencyMillis,
- @Nullable @JsonProperty("lagAggregate") AggregateFunction lagAggregate
+ @Nullable @JsonProperty("lagAggregate") AggregateFunction lagAggregate,
+ @Nullable @JsonProperty("stopTaskCountRatio") Double stopTaskCountRatio
)
{
this.enableTaskAutoScaler = enableTaskAutoScaler != null ? enableTaskAutoScaler : false;
@@ -100,6 +105,12 @@ public class LagBasedAutoScalerConfig implements AutoScalerConfig
this.scaleOutStep = scaleOutStep != null ? scaleOutStep : 2;
this.minTriggerScaleActionFrequencyMillis = minTriggerScaleActionFrequencyMillis
!= null ? minTriggerScaleActionFrequencyMillis : 600000;
+
+ Preconditions.checkArgument(
+ stopTaskCountRatio == null || (stopTaskCountRatio > 0.0 && stopTaskCountRatio <= 1.0),
+ "0.0 < stopTaskCountRatio <= 1.0"
+ );
+ this.stopTaskCountRatio = stopTaskCountRatio;
}
@JsonProperty
@@ -167,7 +178,6 @@ public class LagBasedAutoScalerConfig implements AutoScalerConfig
@Override
@JsonProperty
@Nullable
- @JsonInclude(JsonInclude.Include.NON_NULL)
public Integer getTaskCountStart()
{
return taskCountStart;
@@ -212,6 +222,14 @@ public class LagBasedAutoScalerConfig implements AutoScalerConfig
return lagAggregate;
}
+ @Override
+ @JsonProperty
+ @Nullable
+ public Double getStopTaskCountRatio()
+ {
+ return stopTaskCountRatio;
+ }
+
@Override
public String toString()
{
@@ -232,6 +250,63 @@ public class LagBasedAutoScalerConfig implements AutoScalerConfig
", scaleInStep=" + scaleInStep +
", scaleOutStep=" + scaleOutStep +
", lagAggregate=" + lagAggregate +
+ ", stopTaskCountRatio=" + stopTaskCountRatio +
'}';
}
+
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ LagBasedAutoScalerConfig that = (LagBasedAutoScalerConfig) o;
+
+ return lagCollectionIntervalMillis == that.lagCollectionIntervalMillis &&
+ lagCollectionRangeMillis == that.lagCollectionRangeMillis &&
+ scaleActionStartDelayMillis == that.scaleActionStartDelayMillis &&
+ scaleActionPeriodMillis == that.scaleActionPeriodMillis &&
+ scaleOutThreshold == that.scaleOutThreshold &&
+ scaleInThreshold == that.scaleInThreshold &&
+ Double.compare(that.triggerScaleOutFractionThreshold, triggerScaleOutFractionThreshold) == 0 &&
+ Double.compare(that.triggerScaleInFractionThreshold, triggerScaleInFractionThreshold) == 0 &&
+ taskCountMax == that.taskCountMax &&
+ taskCountMin == that.taskCountMin &&
+ scaleInStep == that.scaleInStep &&
+ scaleOutStep == that.scaleOutStep &&
+ enableTaskAutoScaler == that.enableTaskAutoScaler &&
+ minTriggerScaleActionFrequencyMillis == that.minTriggerScaleActionFrequencyMillis &&
+ Objects.equals(taskCountStart, that.taskCountStart) &&
+ lagAggregate == that.lagAggregate &&
+ Objects.equals(stopTaskCountRatio, that.stopTaskCountRatio);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ lagCollectionIntervalMillis,
+ lagCollectionRangeMillis,
+ scaleActionStartDelayMillis,
+ scaleActionPeriodMillis,
+ scaleOutThreshold,
+ scaleInThreshold,
+ triggerScaleOutFractionThreshold,
+ triggerScaleInFractionThreshold,
+ taskCountMax,
+ taskCountMin,
+ taskCountStart,
+ scaleInStep,
+ scaleOutStep,
+ enableTaskAutoScaler,
+ minTriggerScaleActionFrequencyMillis,
+ lagAggregate,
+ stopTaskCountRatio
+ );
+ }
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/autoscaler/LagBasedAutoScalerConfigTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/autoscaler/LagBasedAutoScalerConfigTest.java
new file mode 100644
index 00000000000..93653f28fe9
--- /dev/null
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/autoscaler/LagBasedAutoScalerConfigTest.java
@@ -0,0 +1,331 @@
+/*
+ * 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.indexing.overlord.supervisor.autoscaler;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.LagBasedAutoScalerConfig;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class LagBasedAutoScalerConfigTest
+{
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ @Test
+ public void testDefaults()
+ {
+ LagBasedAutoScalerConfig config = new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ );
+
+ Assert.assertFalse(config.getEnableTaskAutoScaler());
+ Assert.assertEquals(30000, config.getLagCollectionIntervalMillis());
+ Assert.assertEquals(600000, config.getLagCollectionRangeMillis());
+ Assert.assertEquals(300000, config.getScaleActionStartDelayMillis());
+ Assert.assertEquals(60000, config.getScaleActionPeriodMillis());
+ Assert.assertEquals(6000000, config.getScaleOutThreshold());
+ Assert.assertEquals(1000000, config.getScaleInThreshold());
+ Assert.assertEquals(0.3, config.getTriggerScaleOutFractionThreshold(), 0.00001);
+ Assert.assertEquals(0.9, config.getTriggerScaleInFractionThreshold(), 0.00001);
+ Assert.assertEquals(1, config.getScaleInStep());
+ Assert.assertEquals(2, config.getScaleOutStep());
+ Assert.assertEquals(600000, config.getMinTriggerScaleActionFrequencyMillis());
+ Assert.assertNull(config.getLagAggregate());
+ Assert.assertNull(config.getStopTaskCountRatio());
+ Assert.assertEquals(0, config.getTaskCountMax());
+ Assert.assertEquals(0, config.getTaskCountMin());
+ Assert.assertNull(config.getTaskCountStart());
+ }
+
+ @Test
+ public void testSerde() throws Exception
+ {
+ LagBasedAutoScalerConfig config1 = new LagBasedAutoScalerConfig(
+ 100000L,
+ 100000L,
+ 100000L,
+ 100000L,
+ 1000000L,
+ 100000L,
+ 0.1,
+ 0.9,
+ 10,
+ 5,
+ 1,
+ 1,
+ 5,
+ true,
+ 5000L,
+ AggregateFunction.SUM,
+ 0.1
+ );
+ LagBasedAutoScalerConfig config2 = OBJECT_MAPPER.readValue(
+ OBJECT_MAPPER.writeValueAsString(
+ config1
+ ), LagBasedAutoScalerConfig.class
+ );
+
+ Assert.assertEquals(config1, config2);
+ }
+
+ @Test
+ public void testEnabledTaskCountChecks()
+ {
+ // Should throw if taskCountMax or taskCountMin is missing
+ RuntimeException ex1 = Assert.assertThrows(
+ RuntimeException.class,
+ () ->
+ new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 1,
+ null,
+ null,
+ null,
+ true,
+ null,
+ null,
+ null
+ )
+ );
+ Assert.assertTrue(ex1.getMessage().contains("taskCountMax or taskCountMin can't be null"));
+
+ // Should throw if taskCountMax < taskCountMin
+ RuntimeException ex2 = Assert.assertThrows(
+ RuntimeException.class,
+ () ->
+ new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 2,
+ 1,
+ 3,
+ null,
+ null,
+ true,
+ null,
+ null,
+ null
+ )
+ );
+ Assert.assertTrue(ex2.getMessage().contains("taskCountMax can't lower than taskCountMin"));
+
+ // Should throw if taskCountStart out of range
+ RuntimeException ex3 = Assert.assertThrows(
+ RuntimeException.class,
+ () ->
+ new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 5,
+ 10,
+ 3,
+ null,
+ null,
+ true,
+ null,
+ null,
+ null
+ )
+ );
+ Assert.assertTrue(ex3.getMessage().contains("taskCountMin <= taskCountStart <= taskCountMax"));
+ }
+
+ @Test
+ public void testStopTaskCountRatioBounds()
+ {
+ // Fail if stopTaskCountRatio = 0
+ IllegalArgumentException ex1 = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 10,
+ null,
+ 1,
+ 1,
+ 2,
+ true,
+ null,
+ null,
+ 0.0
+ )
+ );
+ Assert.assertTrue(ex1.getMessage().contains("0.0 < stopTaskCountRatio <= 1.0"));
+
+ // Fail if stopTaskCountRatio < 0
+ IllegalArgumentException ex2 = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 10,
+ null,
+ 1,
+ 1, 2,
+ true,
+ null,
+ null,
+ -0.1
+ )
+ );
+ Assert.assertTrue(ex2.getMessage().contains("0.0 < stopTaskCountRatio <= 1.0"));
+
+ // Fail if stopTaskCountRatio > 1.0
+ IllegalArgumentException ex3 = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 10,
+ null,
+ 1,
+ 1,
+ 2,
+ true,
+ null,
+ null,
+ 1.1
+ )
+ );
+ Assert.assertTrue(ex3.getMessage().contains("0.0 < stopTaskCountRatio <= 1.0"));
+
+ // Should succeed for a valid value
+ LagBasedAutoScalerConfig config = new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 10,
+ null,
+ 1,
+ 1,
+ 2,
+ true,
+ null,
+ null,
+ 0.5
+ );
+ Assert.assertEquals(Double.valueOf(0.5), config.getStopTaskCountRatio());
+ }
+
+ @Test
+ public void testEqualsAndHashCode()
+ {
+ LagBasedAutoScalerConfig config1 = new LagBasedAutoScalerConfig(
+ 1000L,
+ 2000L,
+ 3000L,
+ 4000L,
+ 5000L,
+ 6000L,
+ 0.1,
+ 0.2,
+ 10,
+ 5,
+ 1,
+ 2,
+ 3,
+ true,
+ 7000L,
+ AggregateFunction.SUM,
+ 0.5
+ );
+ LagBasedAutoScalerConfig config2 = new LagBasedAutoScalerConfig(
+ 2000L,
+ 2000L,
+ 5000L,
+ 4500L,
+ 5000L,
+ 6000L,
+ 0.2,
+ 0.4,
+ 40,
+ 20,
+ 1,
+ 4,
+ 1,
+ true,
+ 7000L,
+ AggregateFunction.AVERAGE,
+ 0.5
+ );
+
+ Assert.assertNotEquals(config1, config2);
+ Assert.assertNotEquals(config1.hashCode(), config2.hashCode());
+ }
+}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfigTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfigTest.java
new file mode 100644
index 00000000000..a97e971028c
--- /dev/null
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfigTest.java
@@ -0,0 +1,356 @@
+/*
+ * 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.indexing.seekablestream.supervisor;
+
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.IAE;
+import org.joda.time.Duration;
+import org.joda.time.Period;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class SeekableStreamSupervisorIOConfigTest
+{
+ @Test
+ public void testAllDefaults()
+ {
+ LagAggregator lagAggregator = mock(LagAggregator.class);
+ InputFormat inputFormat = mock(InputFormat.class);
+
+ SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ inputFormat,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ lagAggregator,
+ null,
+ null,
+ null
+ )
+ {
+ };
+
+ Assert.assertEquals("stream", config.getStream());
+ Assert.assertEquals(inputFormat, config.getInputFormat());
+ Assert.assertEquals(Integer.valueOf(1), config.getReplicas());
+ Assert.assertEquals(Integer.valueOf(1), config.getTaskCount());
+ Assert.assertEquals(Duration.standardHours(1), config.getTaskDuration());
+ Assert.assertEquals(Duration.standardSeconds(5), config.getStartDelay());
+ Assert.assertEquals(Duration.standardSeconds(30), config.getPeriod());
+ Assert.assertFalse(config.isUseEarliestSequenceNumber());
+ Assert.assertEquals(Duration.standardMinutes(30), config.getCompletionTimeout());
+ Assert.assertFalse(config.getEarlyMessageRejectionPeriod().isPresent());
+ Assert.assertFalse(config.getLateMessageRejectionPeriod().isPresent());
+ Assert.assertFalse(config.getLateMessageRejectionStartDateTime().isPresent());
+ Assert.assertNull(config.getIdleConfig());
+ Assert.assertNull(config.getStopTaskCount());
+ Assert.assertEquals(lagAggregator, config.getLagAggregator());
+ Assert.assertEquals(1, config.getMaxAllowedStops());
+ }
+
+ @Test
+ public void testAutoScalerEnabledTrueAndFalse()
+ {
+ LagAggregator lagAggregator = mock(LagAggregator.class);
+
+ // autoScalerEnabled = true
+ AutoScalerConfig autoScalerConfig = mock(AutoScalerConfig.class);
+ when(autoScalerConfig.getEnableTaskAutoScaler()).thenReturn(true);
+ when(autoScalerConfig.getTaskCountStart()).thenReturn(5);
+ when(autoScalerConfig.getTaskCountMin()).thenReturn(3);
+
+ SeekableStreamSupervisorIOConfig configAuto = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ 2,
+ 10, // (taskCount should be ignored)
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ autoScalerConfig,
+ lagAggregator,
+ null,
+ null,
+ null
+ )
+ {
+ };
+
+ Assert.assertEquals(Integer.valueOf(5), configAuto.getTaskCount()); // taskCountStart
+
+ // autoScalerEnabled = false
+ SeekableStreamSupervisorIOConfig configNoAuto = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ 2,
+ 10,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ lagAggregator,
+ null,
+ null,
+ null
+ )
+ {
+ };
+
+ Assert.assertEquals(Integer.valueOf(10), configNoAuto.getTaskCount());
+ }
+
+ @Test
+ public void testBothLateMessageRejectionPeriodAndStartDateTime()
+ {
+ LagAggregator lagAggregator = mock(LagAggregator.class);
+
+ IAE ex = Assert.assertThrows(
+ IAE.class,
+ () -> new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ Period.seconds(10),
+ null,
+ null,
+ lagAggregator,
+ DateTimes.nowUtc(),
+ null,
+ null
+ )
+ {
+ }
+ );
+ Assert.assertTrue(
+ ex.getMessage()
+ .contains(
+ "SeekableStreamSupervisorIOConfig does not support both properties lateMessageRejectionStartDateTime and lateMessageRejectionPeriod"
+ )
+ );
+ }
+
+ @Test
+ public void testNullAggregatorThrows()
+ {
+ DruidException ex = Assert.assertThrows(
+ DruidException.class,
+ () -> new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ )
+ {
+ }
+ );
+ Assert.assertTrue(
+ ex.getMessage().contains("'lagAggregator' must be specified in supervisor 'spec.ioConfig'")
+ );
+ }
+
+ @Test
+ public void testGetMaxAllowedStopsScalingDisabled()
+ {
+ LagAggregator lagAggregator = mock(LagAggregator.class);
+
+ // Autoscaler disabled, stopTaskCount unset
+ SeekableStreamSupervisorIOConfig config1 = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ null,
+ 7,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ lagAggregator,
+ null,
+ null,
+ null
+ )
+ {
+ };
+ Assert.assertEquals(7, config1.getMaxAllowedStops());
+
+ // Autoscaler disabled, stopTaskCount set
+ SeekableStreamSupervisorIOConfig config2 = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ null,
+ 7,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ lagAggregator,
+ null,
+ null,
+ 3
+ )
+ {
+ };
+ Assert.assertEquals(3, config2.getMaxAllowedStops());
+ }
+
+ @Test
+ public void testGetMaxAllowedStopsScalingEnabled()
+ {
+ LagAggregator lagAggregator = mock(LagAggregator.class);
+
+ AutoScalerConfig autoScalerConfig = mock(AutoScalerConfig.class);
+
+ // Autoscaler enabled, stopTaskCountRatio set
+ when(autoScalerConfig.getEnableTaskAutoScaler()).thenReturn(true);
+ when(autoScalerConfig.getTaskCountStart()).thenReturn(10);
+ when(autoScalerConfig.getStopTaskCountRatio()).thenReturn(0.5);
+
+ SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ null,
+ 10,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ autoScalerConfig,
+ lagAggregator,
+ null,
+ null,
+ 1
+ )
+ {
+ };
+
+ Assert.assertEquals(5, config.getMaxAllowedStops());
+
+ // Ensure never goes below 1
+ when(autoScalerConfig.getStopTaskCountRatio()).thenReturn(0.05);
+ Assert.assertEquals(1, config.getMaxAllowedStops());
+
+ // Autoscaler enabled, stopTaskCountRatio unset, stopTaskCount set
+ when(autoScalerConfig.getEnableTaskAutoScaler()).thenReturn(true);
+ when(autoScalerConfig.getTaskCountStart()).thenReturn(10);
+ when(autoScalerConfig.getStopTaskCountRatio()).thenReturn(null);
+
+ SeekableStreamSupervisorIOConfig config2 = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ null,
+ 10,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ autoScalerConfig,
+ lagAggregator,
+ null,
+ null,
+ 1
+ )
+ {
+ };
+
+ Assert.assertEquals(1, config2.getMaxAllowedStops());
+
+
+ // Autoscaler enabled, stopTaskCountRatio unset, stopTaskCount unset
+ when(autoScalerConfig.getEnableTaskAutoScaler()).thenReturn(true);
+ when(autoScalerConfig.getTaskCountStart()).thenReturn(10);
+ when(autoScalerConfig.getStopTaskCountRatio()).thenReturn(null);
+
+ SeekableStreamSupervisorIOConfig config3 = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ null,
+ 10,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ autoScalerConfig,
+ lagAggregator,
+ null,
+ null,
+ null
+ )
+ {
+ };
+
+ Assert.assertEquals(10, config3.getMaxAllowedStops());
+ }
+}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
index 54bf3c84bce..e204a67cae8 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
@@ -2634,6 +2634,7 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
null,
true,
null,
+ null,
null
);
SeekableStreamSupervisorIOConfig ioConfig = createSupervisorIOConfig(1, autoScalerConfig);
@@ -2689,6 +2690,55 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
replayAll();
}
+ @Test
+ public void testMaxAllowedStopsWithStopTaskCountRatio()
+ {
+ LagAggregator lagAggregator = createMock(LagAggregator.class);
+ AutoScalerConfig autoScalerConfig = new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 45,
+ null,
+ 5,
+ null,
+ null,
+ true,
+ null,
+ null,
+ 0.4
+ );
+ SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ null,
+ 1,
+ 99,
+ new Period("PT1H"),
+ new Period("PT1S"),
+ new Period("PT30S"),
+ false,
+ new Period("PT30M"),
+ null,
+ null,
+ autoScalerConfig,
+ lagAggregator,
+ null,
+ null,
+ 1 // ensure this is overridden
+ )
+ {
+ };
+
+ Assert.assertEquals(2, config.getMaxAllowedStops());
+ config.setTaskCount(30);
+ Assert.assertEquals(12, config.getMaxAllowedStops());
+ }
+
private static DataSchema getDataSchema()
{
List<DimensionSchema> dimensions = new ArrayList<>();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org