This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch 39.0.0
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/39.0.0 by this push:
new 467c3ce946f minor: Reset auto-scaling cooldown on streaming task
rollover (#20706)
467c3ce946f is described below
commit 467c3ce946f9d74ee714879645fddb964276cb2f
Author: Kashif Faraz <[email protected]>
AuthorDate: Sun Oct 25 20:20:09 2026 +0530
minor: Reset auto-scaling cooldown on streaming task rollover (#20706)
Changes:
- Always consider a scale up or scale down during task rollover
irrespective of the scaling cooldown.
- Since the tasks are being restarted anyway, we might as well choose a
better task count.
- Reset scaling cooldown (driven by `minScaleUpDelay` /
`minScaleDownDelay`) on task rollover.
- Otherwise, if the cooldown elapses soon after the task rollover has
finished, we would go through
another round of task restart.
- Rapid successive task restarts often lead to unnecessary lag buildup
and may even cause task
failures in case of slow segment publish.
- The whole point of the cooldowns is to avoid such task churn.
- Minor cleanup of the `DynamicAllocationTasksNotice`
- Does not affect `lagBased` auto-scaler
(cherry picked from commit fe3c938ec2bba564f26d6af5d26c2edca6b1c953)
---
.../supervisor/SeekableStreamSupervisor.java | 106 +++++++++--------
.../supervisor/autoscaler/CostBasedAutoScaler.java | 2 +-
.../supervisor/autoscaler/LagBasedAutoScaler.java | 8 +-
.../autoscaler/LagBasedAutoScalerConfig.java | 3 +-
.../SeekableStreamSupervisorSpecTest.java | 57 ++++-----
.../SeekableStreamSupervisorStateTest.java | 131 +++++++++++++--------
.../autoscaler/LagBasedAutoScalerTest.java | 5 +-
7 files changed, 165 insertions(+), 147 deletions(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
index f37ecca4ff8..2146713dea2 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
@@ -555,23 +555,30 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
}
- // change taskCount without resubmitting.
+ /**
+ * Notice to scale up or down the number of tasks as dictated by the
auto-scaler.
+ * Scaling takes effect only if there is no task currently pending completion
+ * and the respective scaling cooldown has elapsed. But these restrictions do
+ * not apply if task rollover (i.e. task restart after completion of
taskDuration)
+ * is already in progress.
+ */
private class DynamicAllocationTasksNotice implements Notice
{
- Callable<Integer> computeDesiredTaskCount;
- ServiceEmitter emitter;
- Runnable onSuccessfulScale;
+ final Callable<Integer> computeDesiredTaskCount;
+ final Runnable onSuccessfulScale;
+ final boolean isRolloverInProgress;
+
private static final String TYPE = "dynamic_allocation_tasks_notice";
DynamicAllocationTasksNotice(
Callable<Integer> computeDesiredTaskCount,
Runnable onSuccessfulScale,
- ServiceEmitter emitter
+ boolean isRolloverInProgress
)
{
this.computeDesiredTaskCount = computeDesiredTaskCount;
this.onSuccessfulScale = onSuccessfulScale;
- this.emitter = emitter;
+ this.isRolloverInProgress = isRolloverInProgress;
}
/**
@@ -587,39 +594,33 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
long nowTime = System.currentTimeMillis();
if (spec.isSuspended()) {
log.info(
- "Skipping DynamicAllocationTasksNotice execution because [%s]
supervisor is suspended",
+ "Skipping DynamicAllocationTasksNotice execution because
supervisor[%s] is suspended",
supervisorId
);
return;
}
if (SupervisorStateManager.BasicState.IDLE == getState()) {
log.info(
- "Skipping DynamicAllocationTasksNotice execution because [%s]
supervisor is idle",
+ "Skipping DynamicAllocationTasksNotice execution because
supervisor[%s] is idle",
supervisorId
);
return;
}
- log.debug("PendingCompletionTaskGroups is [%s] for supervisor[%s]
for dataSource[%s]", pendingCompletionTaskGroups,
- supervisorId,
- dataSource
+ log.debug(
+ "PendingCompletionTaskGroups is [%s] for supervisor[%s] for
dataSource[%s]",
+ pendingCompletionTaskGroups, supervisorId, dataSource
);
final int desiredTaskCount = computeDesiredTaskCount.call();
final int currentTaskCount = getCurrentTaskCount();
- ServiceMetricEvent.Builder event = ServiceMetricEvent.builder()
-
.setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
-
.setDimension(DruidMetrics.DATASOURCE, dataSource)
-
.setDimension(DruidMetrics.STREAM, getIoConfig().getStream());
-
if (desiredTaskCount <= 0) {
log.warn(
- "Auto-scaler returned pathological taskCount[%d] for
supervisor[%s] for dataSource[%s]; skipping scale.",
+ "Auto-scaler returned invalid taskCount[%d] for
supervisor[%s], dataSource[%s]; skipping scale.",
desiredTaskCount,
supervisorId,
dataSource
);
- emitter.emit(event.setDimension(AUTOSCALER_SKIP_REASON_DIMENSION,
"Auto-scaler failed to compute a task count")
- .setMetric(AUTOSCALER_REQUIRED_TASKS_METRIC,
desiredTaskCount));
+ emitRequiredTaskCountMetric(desiredTaskCount, "Auto-scaler failed
to compute a task count");
return;
}
@@ -659,8 +660,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
taskCountMin,
taskCountMax
);
- emitter.emit(event.setDimension(AUTOSCALER_SKIP_REASON_DIMENSION,
skipReason)
- .setMetric(AUTOSCALER_REQUIRED_TASKS_METRIC,
desiredTaskCount));
+ emitRequiredTaskCountMetric(desiredTaskCount, skipReason);
return;
}
@@ -669,18 +669,14 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// We want to avoid killing any old generation tasks preemptively,
as that might cause the current generation tasks' offsets to become invalid.
for (CopyOnWriteArrayList<TaskGroup> list :
pendingCompletionTaskGroups.values()) {
// There are expected to be pending tasks if this scaling is
happening on task rollover
- if (!list.isEmpty() && !isScalingTasksOnRollover.get()) {
+ if (!list.isEmpty() && !isRolloverInProgress) {
log.info(
"Skipping DynamicAllocationTasksNotice execution for
supervisor[%s] for datasource[%s] because following tasks are pending [%s]",
supervisorId,
dataSource,
list
);
- emitter.emit(event.setDimension(
- AUTOSCALER_SKIP_REASON_DIMENSION,
- "There are tasks pending completion"
- )
- .setMetric(AUTOSCALER_REQUIRED_TASKS_METRIC,
desiredTaskCount));
+ emitRequiredTaskCountMetric(desiredTaskCount, "There are tasks
pending completion");
return;
}
}
@@ -702,7 +698,9 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
cooldownMillis =
autoScalerConfig.getMinScaleDownDelay().getMillis();
}
- if (nowTime - dynamicTriggerLastScaleRunTime < cooldownMillis) {
+ // Ignore scaling cooldown on task rollover. Since tasks are being
+ // restarted anyway, we might as well consider changing the
taskCount.
+ if (!isRolloverInProgress && (nowTime -
dynamicTriggerLastScaleRunTime) < cooldownMillis) {
log.info(
"DynamicAllocationTasksNotice submitted again in [%d]ms, [%s]
cooldown is [%d]ms for supervisor[%s] for dataSource[%s], skipping it! scaler
wants [%d] (clamped [%d]), current task count is [%d]",
nowTime - dynamicTriggerLastScaleRunTime,
@@ -715,16 +713,12 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
currentTaskCount
);
- emitter.emit(event.setDimension(
- AUTOSCALER_SKIP_REASON_DIMENSION,
- "Scale cooldown not elapsed yet"
- )
- .setMetric(AUTOSCALER_REQUIRED_TASKS_METRIC,
desiredTaskCount));
+ emitRequiredTaskCountMetric(desiredTaskCount, "Scale cooldown not
elapsed yet");
return;
}
// Emit the scaler's unclamped preferred count so operators see what
it wants.
- emitter.emit(event.setMetric(AUTOSCALER_REQUIRED_TASKS_METRIC,
desiredTaskCount));
+ emitRequiredTaskCountMetric(desiredTaskCount, null);
boolean allocationSuccess = changeTaskCount(clampedTaskCount);
if (allocationSuccess) {
@@ -738,6 +732,20 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
}
+ private void emitRequiredTaskCountMetric(int desiredTaskCount, @Nullable
String skipReason)
+ {
+ final ServiceMetricEvent.Builder event = ServiceMetricEvent
+ .builder()
+ .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
+ .setDimension(DruidMetrics.DATASOURCE, dataSource)
+ .setDimension(DruidMetrics.STREAM, getIoConfig().getStream());
+ if (skipReason != null) {
+ event.setDimension(AUTOSCALER_SKIP_REASON_DIMENSION, skipReason);
+ }
+ event.setMetric(AUTOSCALER_REQUIRED_TASKS_METRIC, desiredTaskCount);
+ emitter.emit(event);
+ }
+
@Override
public String getType()
{
@@ -1159,7 +1167,6 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
* Wired by {@link SupervisorManager} after supervisor creation.
*/
private volatile SupervisorTaskAutoScaler taskAutoScaler;
- private final AtomicBoolean isScalingTasksOnRollover = new
AtomicBoolean(false);
// snapshots latest sequences from the stream to be verified in the next run
cycle of inactive stream check
private final Map<PartitionIdType, SequenceOffsetType>
previousSequencesFromStream = new HashMap<>();
@@ -1689,21 +1696,20 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
public Runnable buildDynamicAllocationTask(
Callable<Integer> scaleAction,
- Runnable onSuccessfulScale,
- ServiceEmitter emitter
+ Runnable onSuccessfulScale
)
{
- return () -> addNotice(new DynamicAllocationTasksNotice(scaleAction,
onSuccessfulScale, emitter));
+ return () -> addNotice(new DynamicAllocationTasksNotice(scaleAction,
onSuccessfulScale, false));
}
@VisibleForTesting
void handleDynamicAllocationTasksNotice(
Callable<Integer> scaleAction,
Runnable onSuccessfulScale,
- ServiceEmitter emitter
+ boolean isRolloverInProgress
)
{
- new DynamicAllocationTasksNotice(scaleAction, onSuccessfulScale,
emitter).handle();
+ new DynamicAllocationTasksNotice(scaleAction, onSuccessfulScale,
isRolloverInProgress).handle();
}
private Runnable buildRunTask()
@@ -4050,17 +4056,15 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
"Updating taskCount for supervisor[%s] from [%d] to [%d] during
rollover.",
supervisorId, currentTaskCount, rolloverTaskCount
);
- isScalingTasksOnRollover.set(true);
- try {
- new DynamicAllocationTasksNotice(
- () -> rolloverTaskCount,
- () -> {},
- emitter
- ).handle();
- }
- finally {
- isScalingTasksOnRollover.set(false);
- }
+ handleDynamicAllocationTasksNotice(
+ () -> rolloverTaskCount,
+ () -> {},
+ true
+ );
+ } else if (rolloverTaskCount != SupervisorTaskAutoScaler.CANNOT_COMPUTE)
{
+ // Reset last scale time even if taskCount is not being changed.
+ // Otherwise, we might have a scaling event right after the task
rollover.
+ dynamicTriggerLastScaleRunTime = System.currentTimeMillis();
}
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
index 455ecb8e099..e34ad386099 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
@@ -174,7 +174,7 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
public void start()
{
autoscalerExecutor.scheduleAtFixedRate(
-
supervisor.buildDynamicAllocationTask(this::computeTaskCountForScaleAction, ()
-> {}, emitter),
+
supervisor.buildDynamicAllocationTask(this::computeTaskCountForScaleAction, ()
-> {}),
config.getScaleActionPeriodMillis(),
config.getScaleActionPeriodMillis(),
TimeUnit.MILLISECONDS
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java
index adb041911af..a1ff7b004e7 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java
@@ -30,7 +30,6 @@ import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervi
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.utils.CollectionUtils;
import java.util.ArrayList;
@@ -49,15 +48,13 @@ public class LagBasedAutoScaler implements
SupervisorTaskAutoScaler
private final SupervisorSpec spec;
private final SeekableStreamSupervisor supervisor;
private final LagBasedAutoScalerConfig lagBasedAutoScalerConfig;
- private final ServiceEmitter emitter;
private static final ReentrantLock LOCK = new ReentrantLock(true);
public LagBasedAutoScaler(
SeekableStreamSupervisor supervisor,
LagBasedAutoScalerConfig autoScalerConfig,
- SupervisorSpec spec,
- ServiceEmitter emitter
+ SupervisorSpec spec
)
{
this.lagBasedAutoScalerConfig = autoScalerConfig;
@@ -73,7 +70,6 @@ public class LagBasedAutoScaler implements
SupervisorTaskAutoScaler
this.lagComputationExec =
Execs.scheduledSingleThreaded(StringUtils.encodeForFormat(supervisorId) +
"-Computation-%d");
this.spec = spec;
this.supervisor = supervisor;
- this.emitter = emitter;
}
@Override
@@ -114,7 +110,7 @@ public class LagBasedAutoScaler implements
SupervisorTaskAutoScaler
TimeUnit.MILLISECONDS
);
allocationExec.scheduleAtFixedRate(
- supervisor.buildDynamicAllocationTask(scaleAction, onSuccessfulScale,
emitter),
+ supervisor.buildDynamicAllocationTask(scaleAction, onSuccessfulScale),
lagBasedAutoScalerConfig.getScaleActionStartDelayMillis() +
lagBasedAutoScalerConfig
.getLagCollectionRangeMillis(),
lagBasedAutoScalerConfig.getScaleActionPeriodMillis(),
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 1ec08384a40..a2800ebc7c4 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
@@ -205,8 +205,7 @@ public class LagBasedAutoScalerConfig implements
AutoScalerConfig
return new LagBasedAutoScaler(
(SeekableStreamSupervisor) supervisor,
this,
- spec,
- emitter
+ spec
);
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
index ae6f3a846c0..80b6b722de7 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
@@ -42,7 +42,6 @@ import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.parsers.JSONPathSpec;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.metrics.DruidMonitorSchedulerConfig;
-import org.apache.druid.java.util.metrics.StubServiceEmitter;
import org.apache.druid.metadata.MetadataSupervisorManager;
import org.apache.druid.metadata.TestSupervisorSpec;
import org.apache.druid.query.DruidMetrics;
@@ -54,7 +53,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
-
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -426,8 +424,6 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent()).anyTimes();
EasyMock.replay(taskMaster);
- StubServiceEmitter dynamicActionEmitter = new StubServiceEmitter();
-
TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor(10);
LagBasedAutoScaler autoScaler = new LagBasedAutoScaler(
@@ -436,8 +432,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleOutProperties(10),
LagBasedAutoScalerConfig.class
),
- spec,
- dynamicActionEmitter
+ spec
);
supervisor.start();
autoScaler.start();
@@ -449,12 +444,13 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
int taskCountAfterScaleOut = supervisor.getIoConfig().getTaskCount();
Assertions.assertEquals(3, taskCountAfterScaleOut);
Assertions.assertTrue(
- dynamicActionEmitter
+ emitter
.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC)
.stream()
.map(metric ->
metric.getUserDims().get(SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION))
.filter(Objects::nonNull)
- .anyMatch("Scale cooldown not elapsed yet"::equals));
+ .anyMatch("Scale cooldown not elapsed yet"::equals)
+ );
emitter.verifyEmitted(SeekableStreamSupervisor.AUTOSCALER_SCALING_TIME_METRIC,
1);
autoScaler.reset();
autoScaler.stop();
@@ -482,7 +478,6 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent()).anyTimes();
EasyMock.replay(taskMaster);
- StubServiceEmitter dynamicActionEmitter = new StubServiceEmitter();
TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor(10);
LagBasedAutoScaler autoScaler = new LagBasedAutoScaler(
@@ -491,8 +486,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleOutProperties(2),
LagBasedAutoScalerConfig.class
),
- spec,
- dynamicActionEmitter
+ spec
);
supervisor.getIoConfig().setTaskCount(2);
supervisor.start();
@@ -501,13 +495,13 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
Thread.sleep(1000);
Assertions.assertTrue(
- dynamicActionEmitter
+ emitter
.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC)
.stream()
.map(metric ->
metric.getUserDims().get(SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION))
.filter(Objects::nonNull)
- .anyMatch("Already at max task count"::equals));
-
emitter.verifyNotEmitted(SeekableStreamSupervisor.AUTOSCALER_SCALING_TIME_METRIC);
+ .anyMatch("Already at max task count"::equals)
+ );
autoScaler.reset();
autoScaler.stop();
@@ -546,8 +540,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleOutProperties(2),
LagBasedAutoScalerConfig.class
),
- spec,
- emitter
+ spec
);
supervisor.start();
autoScaler.start();
@@ -591,8 +584,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleOutProperties(3),
LagBasedAutoScalerConfig.class
),
- spec,
- emitter
+ spec
);
supervisor.start();
autoScaler.start();
@@ -638,8 +630,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleInProperties(),
LagBasedAutoScalerConfig.class
),
- spec,
- emitter
+ spec
);
// enable autoscaler so that taskcount config will be ignored and init
value of taskCount will use taskCountMin.
@@ -713,8 +704,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
misconfiguredProps,
LagBasedAutoScalerConfig.class
),
- spec,
- emitter
+ spec
);
// Initial taskCount comes from the ioConfig's
autoScalerConfig.taskCountMin (15) since
@@ -775,7 +765,6 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent()).anyTimes();
EasyMock.replay(taskMaster);
- StubServiceEmitter dynamicActionEmitter = new StubServiceEmitter();
TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor(10)
{
@Override
@@ -791,8 +780,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
scaleInProps,
LagBasedAutoScalerConfig.class
),
- spec,
- dynamicActionEmitter
+ spec
);
supervisor.getIoConfig().setTaskCount(2);
supervisor.start();
@@ -801,13 +789,13 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
Thread.sleep(1000);
Assertions.assertTrue(
- dynamicActionEmitter
+ emitter
.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC)
.stream()
.map(metric ->
metric.getUserDims().get(SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION))
.filter(Objects::nonNull)
- .anyMatch("Already at min task count"::equals));
-
emitter.verifyNotEmitted(SeekableStreamSupervisor.AUTOSCALER_SCALING_TIME_METRIC);
+ .anyMatch("Already at min task count"::equals)
+ );
autoScaler.reset();
autoScaler.stop();
@@ -1219,7 +1207,6 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent()).anyTimes();
EasyMock.replay(taskMaster);
- StubServiceEmitter dynamicActionEmitter = new StubServiceEmitter();
TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor(10);
LagBasedAutoScaler autoScaler = new LagBasedAutoScaler(
@@ -1228,8 +1215,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleOutProperties(2),
LagBasedAutoScalerConfig.class
),
- spec,
- dynamicActionEmitter
+ spec
);
supervisor.addTaskGroupToPendingCompletionTaskGroup(
@@ -1250,7 +1236,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
Assertions.assertEquals(1, supervisor.getIoConfig().getTaskCount());
Assertions.assertTrue(
- dynamicActionEmitter
+ emitter
.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC)
.stream()
.map(metric ->
metric.getUserDims().get(SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION))
@@ -1258,7 +1244,6 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
.anyMatch("There are tasks pending completion"::equals)
);
-
emitter.verifyNotEmitted(SeekableStreamSupervisor.AUTOSCALER_SCALING_TIME_METRIC);
autoScaler.reset();
autoScaler.stop();
}
@@ -1293,8 +1278,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleOutProperties(2),
LagBasedAutoScalerConfig.class
),
- spec,
- emitter
+ spec
);
supervisor.start();
@@ -1346,8 +1330,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
getScaleOutProperties(2),
LagBasedAutoScalerConfig.class
),
- spec,
- emitter
+ spec
);
supervisor.start();
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 912455f8ba2..b7177f931e9 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
@@ -4214,17 +4214,17 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
final long zeroCooldown = 0L;
final long unusedCooldown = Duration.standardHours(1).getMillis();
- final StubServiceEmitter scalingEmitter =
setupSupervisorForAutoScalingTest(zeroCooldown, unusedCooldown, 2);
+ setupSupervisorForAutoScalingTest(zeroCooldown, unusedCooldown, 2);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
// minScaleUpDelay = 0 means any scale-up is immediately allowed.
- supervisor.handleDynamicAllocationTasksNotice(() -> 5, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 5, () -> {}, false);
Assertions.assertEquals(5, supervisor.getIoConfig().getTaskCount());
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(1, events.size(), "Exactly one required-tasks
emission expected");
assertScaledToTaskCount(events.get(0), 5);
}
@@ -4234,20 +4234,20 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
final long scaleUpCooldown = Duration.standardHours(1).getMillis();
final long unusedCooldown = 0L;
- final StubServiceEmitter scalingEmitter =
setupSupervisorForAutoScalingTest(scaleUpCooldown, unusedCooldown, 2);
+ setupSupervisorForAutoScalingTest(scaleUpCooldown, unusedCooldown, 2);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
// First scale-up succeeds and stamps the last-scale timestamp.
- supervisor.handleDynamicAllocationTasksNotice(() -> 5, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 5, () -> {}, false);
Assertions.assertEquals(5, supervisor.getIoConfig().getTaskCount());
// Second scale-up is within the 1h minScaleUpDelay window and must be
blocked.
- supervisor.handleDynamicAllocationTasksNotice(() -> 7, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 7, () -> {}, false);
Assertions.assertEquals(5, supervisor.getIoConfig().getTaskCount(),
"Second scale-up must not take effect");
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(2, events.size(), "Two required-tasks emissions
expected (one applied, one skipped)");
// First emission: the successful scale carries no skip-reason dim and
reports the applied count.
assertScaledToTaskCount(events.get(0), 5);
@@ -4255,22 +4255,46 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
assertScaleSkipped(events.get(1), 7, "Scale cooldown not elapsed yet");
}
+ @Test
+ public void
test_dynamicAllocation_allowsScaleUpDuringRollover_evenIfCooldownNotElapsed()
+ {
+ final long scaleUpCooldown = Duration.standardHours(1).getMillis();
+ final long unusedCooldown = Duration.standardHours(1).getMillis();
+ setupSupervisorForAutoScalingTest(scaleUpCooldown, unusedCooldown, 2);
+ final TestSeekableStreamSupervisor supervisor =
+ new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
+
+ // First scale-up succeeds and stamps the last-scale timestamp.
+ supervisor.handleDynamicAllocationTasksNotice(() -> 5, () -> {}, false);
+ Assertions.assertEquals(5, supervisor.getIoConfig().getTaskCount());
+
+ // Second scale-up also succeeds since cooldown is ignored on rollover
+ supervisor.handleDynamicAllocationTasksNotice(() -> 10, () -> {}, true);
+ Assertions.assertEquals(10, supervisor.getIoConfig().getTaskCount());
+
+ final List<ServiceMetricEvent> events =
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+ Assertions.assertEquals(2, events.size(), "Exactly one required-tasks
emission expected");
+ assertScaledToTaskCount(events.get(0), 5);
+ assertScaledToTaskCount(events.get(1), 10);
+ }
+
@Test
public void testDynamicAllocationScaleDownAllowedWhenCooldownElapsed()
{
final long unusedCooldown = Duration.standardHours(1).getMillis();
final long zeroCooldown = 0L;
- final StubServiceEmitter scalingEmitter =
setupSupervisorForAutoScalingTest(unusedCooldown, zeroCooldown, 5);
+ setupSupervisorForAutoScalingTest(unusedCooldown, zeroCooldown, 5);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
// minScaleDownDelay = 0 means any scale-down is immediately allowed.
- supervisor.handleDynamicAllocationTasksNotice(() -> 2, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 2, () -> {}, false);
Assertions.assertEquals(2, supervisor.getIoConfig().getTaskCount());
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(1, events.size(), "Exactly one required-tasks
emission expected");
assertScaledToTaskCount(events.get(0), 2);
}
@@ -4280,41 +4304,64 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
final long unusedCooldown = 0L;
final long scaleDownCooldown = Duration.standardHours(1).getMillis();
- final StubServiceEmitter scalingEmitter =
setupSupervisorForAutoScalingTest(unusedCooldown, scaleDownCooldown, 5);
+ setupSupervisorForAutoScalingTest(unusedCooldown, scaleDownCooldown, 5);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
// First scale-down succeeds and stamps the last-scale timestamp.
- supervisor.handleDynamicAllocationTasksNotice(() -> 3, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 3, () -> {}, false);
Assertions.assertEquals(3, supervisor.getIoConfig().getTaskCount());
// Second scale-down is within the 1h minScaleDownDelay window and must be
blocked.
- supervisor.handleDynamicAllocationTasksNotice(() -> 1, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 1, () -> {}, false);
Assertions.assertEquals(3, supervisor.getIoConfig().getTaskCount(),
"Second scale-down must not take effect");
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(2, events.size(), "Two required-tasks emissions
expected (one applied, one skipped)");
assertScaledToTaskCount(events.get(0), 3);
assertScaleSkipped(events.get(1), 1, "Scale cooldown not elapsed yet");
}
+ @Test
+ public void
test_dynamicAllocation_allowsScaleDownDuringRollover_evenIfCooldownNotElapsed()
+ {
+ final long unusedCooldown = Duration.standardHours(1).getMillis();
+ final long scaleDownCooldown = Duration.standardHours(1).getMillis();
+ setupSupervisorForAutoScalingTest(unusedCooldown, scaleDownCooldown, 5);
+ final TestSeekableStreamSupervisor supervisor =
+ new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
+
+ // First scale-down succeeds and stamps the last-scale timestamp.
+ supervisor.handleDynamicAllocationTasksNotice(() -> 3, () -> {}, false);
+ Assertions.assertEquals(3, supervisor.getIoConfig().getTaskCount());
+
+ // Second scale down also passes since cooldown is ignored
+ supervisor.handleDynamicAllocationTasksNotice(() -> 1, () -> {}, true);
+ Assertions.assertEquals(1, supervisor.getIoConfig().getTaskCount());
+
+ final List<ServiceMetricEvent> events =
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+ Assertions.assertEquals(2, events.size(), "Exactly one required-tasks
emission expected");
+ assertScaledToTaskCount(events.get(0), 3);
+ assertScaledToTaskCount(events.get(1), 1);
+ }
+
@Test
public void testDynamicAllocationClampsDesiredAboveMaxToMax()
{
// Scaler returns 20, but taskCountMax=10. Supervisor must clamp and scale
to 10.
- final StubServiceEmitter scalingEmitter =
- setupSupervisorForAutoScalingTest(0L, 0L, 3, 1, 10);
+ setupSupervisorForAutoScalingTest(0L, 0L, 3, 1, 10);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
- supervisor.handleDynamicAllocationTasksNotice(() -> 20, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 20, () -> {}, false);
// Task count is clamped to max (10), not the scaler's desired (20).
Assertions.assertEquals(10, supervisor.getIoConfig().getTaskCount());
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(1, events.size());
// The emitted metric value reflects the scaler's unclamped desired (the
operator hint), not
// the clamped value the supervisor actually applied.
@@ -4325,17 +4372,16 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
public void testDynamicAllocationClampsDesiredBelowMinToMin()
{
// Scaler returns 1, but taskCountMin=3. Supervisor must clamp and scale
to 3.
- final StubServiceEmitter scalingEmitter =
- setupSupervisorForAutoScalingTest(0L, 0L, 5, 3, 10);
+ setupSupervisorForAutoScalingTest(0L, 0L, 5, 3, 10);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
- supervisor.handleDynamicAllocationTasksNotice(() -> 1, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 1, () -> {}, false);
Assertions.assertEquals(3, supervisor.getIoConfig().getTaskCount());
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(1, events.size());
assertScaledToTaskCount(events.get(0), 1);
}
@@ -4345,17 +4391,16 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
// Current (10) is already at configured max (10). Scaler wants 15 (above
max). Supervisor
// clamps to 10 which equals current -> emits "Already at max task count"
skip reason.
- final StubServiceEmitter scalingEmitter =
- setupSupervisorForAutoScalingTest(0L, 0L, 10, 1, 10);
+ setupSupervisorForAutoScalingTest(0L, 0L, 10, 1, 10);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
- supervisor.handleDynamicAllocationTasksNotice(() -> 15, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 15, () -> {}, false);
Assertions.assertEquals(10, supervisor.getIoConfig().getTaskCount(), "Task
count must not change when at max");
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(1, events.size());
assertScaleSkipped(events.get(0), 15, "Already at max task count");
}
@@ -4365,17 +4410,16 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
// Current (3) is already at configured min (3). Scaler wants 1 (below
min). Supervisor clamps
// to 3 which equals current -> emits "Already at min task count" skip
reason.
- final StubServiceEmitter scalingEmitter =
- setupSupervisorForAutoScalingTest(0L, 0L, 3, 3, 10);
+ setupSupervisorForAutoScalingTest(0L, 0L, 3, 3, 10);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
- supervisor.handleDynamicAllocationTasksNotice(() -> 1, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 1, () -> {}, false);
Assertions.assertEquals(3, supervisor.getIoConfig().getTaskCount(), "Task
count must not change when at min");
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(1, events.size());
assertScaleSkipped(events.get(0), 1, "Already at min task count");
}
@@ -4384,17 +4428,16 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
public void testDynamicAllocationEmitsNothingWhenDesiredEqualsCurrent()
{
// No skip metric in the steady-state no-op (avoid emitting on every tick).
- final StubServiceEmitter scalingEmitter =
- setupSupervisorForAutoScalingTest(0L, 0L, 5, 1, 10);
+ setupSupervisorForAutoScalingTest(0L, 0L, 5, 1, 10);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
- supervisor.handleDynamicAllocationTasksNotice(() -> 5, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> 5, () -> {}, false);
Assertions.assertEquals(5, supervisor.getIoConfig().getTaskCount());
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertTrue(events.isEmpty(), "No metric should be emitted in
the steady-state no-op case");
}
@@ -4403,17 +4446,16 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
// Scaler contract: a non-positive return means "I could not compute a
useful answer".
// Supervisor must not scale and must emit a skip reason surfacing the
scaler's failure.
- final StubServiceEmitter scalingEmitter =
- setupSupervisorForAutoScalingTest(0L, 0L, 5, 1, 10);
+ setupSupervisorForAutoScalingTest(0L, 0L, 5, 1, 10);
final TestSeekableStreamSupervisor supervisor =
new
StateOverrideTestSeekableStreamSupervisor(SupervisorStateManager.BasicState.RUNNING);
- supervisor.handleDynamicAllocationTasksNotice(() -> -1, () -> {},
scalingEmitter);
+ supervisor.handleDynamicAllocationTasksNotice(() -> -1, () -> {}, false);
Assertions.assertEquals(5, supervisor.getIoConfig().getTaskCount(), "Task
count must not change on pathological return");
final List<ServiceMetricEvent> events =
-
scalingEmitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
+
emitter.getMetricEvents(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC);
Assertions.assertEquals(1, events.size());
assertScaleSkipped(events.get(0), -1, "Auto-scaler failed to compute a
task count");
}
@@ -4479,16 +4521,16 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
* without hitting unmocked calls. Returns a dedicated emitter for the
caller to pass into the
* notice handler so dynamic-allocation events can be asserted in isolation.
*/
- private StubServiceEmitter setupSupervisorForAutoScalingTest(
+ private void setupSupervisorForAutoScalingTest(
long minScaleUpDelayMillis,
long minScaleDownDelayMillis,
int initialTaskCount
)
{
- return setupSupervisorForAutoScalingTest(minScaleUpDelayMillis,
minScaleDownDelayMillis, initialTaskCount, 1, 100);
+ setupSupervisorForAutoScalingTest(minScaleUpDelayMillis,
minScaleDownDelayMillis, initialTaskCount, 1, 100);
}
- private StubServiceEmitter setupSupervisorForAutoScalingTest(
+ private void setupSupervisorForAutoScalingTest(
long minScaleUpDelayMillis,
long minScaleDownDelayMillis,
int initialTaskCount,
@@ -4507,7 +4549,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
autoScalerConfig,
null
);
- return resetSpecAndTaskMasterForScaling(ioConfig);
+ resetSpecAndTaskMasterForScaling(ioConfig);
}
/**
@@ -4587,10 +4629,8 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
};
}
- private StubServiceEmitter
resetSpecAndTaskMasterForScaling(SeekableStreamSupervisorIOConfig ioConfig)
+ private void
resetSpecAndTaskMasterForScaling(SeekableStreamSupervisorIOConfig ioConfig)
{
- final StubServiceEmitter scalingEmitter = new
StubServiceEmitter("scaling", "localhost");
-
EasyMock.reset(spec, taskMaster);
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
@@ -4607,6 +4647,5 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent()).anyTimes();
replayAll();
- return scalingEmitter;
}
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerTest.java
index 69aa401ea82..0f745b724af 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerTest.java
@@ -22,7 +22,6 @@ package
org.apache.druid.indexing.seekablestream.supervisor.autoscaler;
import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIOConfig;
-import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -42,7 +41,6 @@ public class LagBasedAutoScalerTest
private SupervisorSpec mockSpec;
private SeekableStreamSupervisor mockSupervisor;
private SeekableStreamSupervisorIOConfig mockIoConfig;
- private ServiceEmitter mockEmitter;
private LagBasedAutoScalerConfig config;
@BeforeEach
@@ -50,7 +48,6 @@ public class LagBasedAutoScalerTest
{
mockSpec = Mockito.mock(SupervisorSpec.class);
mockSupervisor = Mockito.mock(SeekableStreamSupervisor.class);
- mockEmitter = Mockito.mock(ServiceEmitter.class);
mockIoConfig = Mockito.mock(SeekableStreamSupervisorIOConfig.class);
when(mockSpec.getId()).thenReturn("test-supervisor");
@@ -121,7 +118,7 @@ public class LagBasedAutoScalerTest
private LagBasedAutoScaler createAutoScaler()
{
- return new LagBasedAutoScaler(mockSupervisor, config, mockSpec,
mockEmitter);
+ return new LagBasedAutoScaler(mockSupervisor, config, mockSpec);
}
private List<Long> createLagSamples(long lag)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
