You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by ah...@apache.org on 2026/09/23 21:00:02 UTC
(druid) branch master updated: fix: stop replica top-up churn for completed bounded task groups (#20387)
This is an automated email from the ASF dual-hosted git repository.
aho135 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 294fe36329e fix: stop replica top-up churn for completed bounded task groups (#20387)
294fe36329e is described below
commit 294fe36329e6b4882a33d39f4e73f6779873f0c4
Author: aho135 <an...@gmail.com>
AuthorDate: Wed Sep 23 13:59:55 2026 -0700
fix: stop replica top-up churn for completed bounded task groups (#20387)
* fix(kafka,kinesis): stop replica top-up churn for completed bounded task groups
In bounded (backfill) mode with replicas > 1, once a task group's committed
offsets reached the configured end, exiting replicas dropped the group below
the configured replica count. The replica top-up loop in createNewTasks() then
kept creating replacement replicas whose start offset was already at the bounded
end, so they completed instantly and re-triggered the top-up, churning tasks
endlessly.
Guard the top-up loop with hasTaskGroupReachedBoundedEnd(), mirroring the
existing guard on the task-group recreation path in the same method.
* Address review: log bounded top-up skip at info level
* Move bounded top-up-guard test to indexing-service and make it fail without the fix
The prior KafkaSupervisorTest case passed even without the guard (it never
reached the replica top-up loop). Replace it with a SeekableStreamSupervisorStateTest
case that flows through runInternal(): it injects an actively-reading bounded task
group that has reached its end offsets with fewer tasks than replicas, and asserts
no replacement replica is submitted. Verified this fails without the guard. This also
provides in-module coverage of the new code path.
* Address review: only check bounded-end inside the replica top-up branch
Move the hasTaskGroupReachedBoundedEnd() guard inside the
replicas > tasks.size() branch so the metadata-store lookup runs only when a
top-up would otherwise be issued, rather than for every active group on every
supervisor run.
* Address review: exercise the metadata completion path in the top-up guard test
Use a non-empty configured bounded range [0, 100) and a metadata store reporting
committed offsets that have reached the end plus a matching bounded config, so the
test drives hasTaskGroupReachedBoundedEnd()'s metadata-offset path rather than the
empty-range short-circuit. Verified it still fails without the guard.
* Address review: trim redundant javadoc on the top-up guard test
---
.../supervisor/SeekableStreamSupervisor.java | 13 +++
.../SeekableStreamSupervisorStateTest.java | 124 +++++++++++++++++++++
2 files changed, 137 insertions(+)
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 6efd24d4b09..ab204a63acf 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
@@ -4489,6 +4489,19 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
if (ioConfig.getReplicas() > taskGroup.tasks.size()) {
+ // In bounded mode, do not top up replicas for a task group that has already reached its end offsets.
+ // Otherwise, as completed replicas exit, this loop keeps recreating replacement replicas whose start
+ // offsets are already at the bounded end, producing an endless churn of tasks that complete instantly.
+ // This mirrors the completion guard on the task-group recreation path in this method. The bounded-end
+ // check (a metadata-store lookup) is intentionally kept inside this branch so it only runs when a top-up
+ // would otherwise happen, not for every active group on every run.
+ if (ioConfig.isBounded() && hasTaskGroupReachedBoundedEnd(groupId)) {
+ log.info(
+ "Bounded taskGroup[%d] has reached end offsets, skipping replica top-up",
+ groupId
+ );
+ continue;
+ }
log.info(
"Number of tasks[%d] does not match configured numReplicas[%d] in taskGroup[%d], creating more tasks.",
taskGroup.tasks.size(), ioConfig.getReplicas(), groupId
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 540b58049b9..7aace3d7540 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
@@ -2961,6 +2961,122 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
verifyAll();
}
+ /**
+ * A bounded task group that has reached its configured end offsets is not topped up with replacement replicas,
+ * exercising the metadata-based completion path (non-empty range with committed offsets at the end).
+ */
+ @Test
+ public void testCreateNewTasks_boundedGroupReachedEnd_doesNotTopUpReplicas()
+ {
+ // replicas = 2, taskCount = 1, non-empty bounded range [0, 100).
+ final BoundedStreamConfig boundedConfig = new BoundedStreamConfig(
+ ImmutableMap.of("0", "0"),
+ ImmutableMap.of("0", "100")
+ );
+ final SeekableStreamSupervisorIOConfig ioConfig =
+ new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(2)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withBoundedStreamConfig(boundedConfig)
+ .build();
+
+ // A single already-running task for group 0 whose committed offset ("100") has reached the configured end, so the
+ // group has one task, i.e. fewer than the configured replica count (2). That would normally trigger a top-up.
+ // minMsgTime/maxMsgTime and start sequences must match the injected task group so isTaskCurrent() keeps the task.
+ final SeekableStreamIndexTaskIOConfig taskIoConfig = createTaskIoConfigExt(
+ 0,
+ Map.of("0", "100"),
+ Map.of("0", "100"),
+ "test",
+ null,
+ null,
+ Set.of(),
+ ioConfig
+ );
+ final TestSeekableStreamIndexTask task1 = createTestTask("task1", "0", null, taskIoConfig, recordSupplier);
+
+ EasyMock.reset(spec);
+ EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
+ EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
+ EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
+ EasyMock.expect(spec.getIoConfig()).andReturn(ioConfig).anyTimes();
+ EasyMock.expect(spec.getTuningConfig()).andReturn(getTuningConfig()).anyTimes();
+ EasyMock.expect(spec.getEmitter()).andReturn(emitter).anyTimes();
+ EasyMock.expect(spec.getContextValue(DruidMetrics.TAGS)).andReturn(METRIC_TAGS).anyTimes();
+ EasyMock.expect(spec.isSuspended()).andReturn(false).anyTimes();
+
+ // Metadata store reports committed offsets that have reached the end, with a bounded config matching the
+ // supervisor's, so hasTaskGroupReachedBoundedEnd() returns true through its metadata-offset path.
+ EasyMock.reset(indexerMetadataStorageCoordinator);
+ EasyMock.expect(indexerMetadataStorageCoordinator.retrieveDataSourceMetadata(SUPERVISOR_ID))
+ .andReturn(new TestSeekableStreamDataSourceMetadata(
+ new SeekableStreamEndSequenceNumbers<>(STREAM, ImmutableMap.of("0", "100")),
+ boundedConfig
+ ))
+ .anyTimes();
+
+ EasyMock.expect(recordSupplier.getPartitionIds(STREAM)).andReturn(ImmutableSet.of("0")).anyTimes();
+ EasyMock.expect(taskStorage.getStatus("task1")).andReturn(Optional.of(TaskStatus.running("task1"))).anyTimes();
+ EasyMock.expect(taskStorage.getTask("task1")).andReturn(Optional.of(task1)).anyTimes();
+ EasyMock.expect(taskQueue.getActiveTasksForDatasource(DATASOURCE))
+ .andReturn(Map.of(task1.getId(), task1))
+ .anyTimes();
+ // Task checkpoints match the reached-end committed offsets so the task is not killed as inconsistent.
+ final TreeMap<Integer, Map<String, String>> task1Checkpoints = new TreeMap<>();
+ task1Checkpoints.put(0, ImmutableMap.of("0", "100"));
+ EasyMock.expect(indexTaskClient.getCheckpointsAsync(EasyMock.anyString(), EasyMock.anyBoolean()))
+ .andReturn(Futures.immediateFuture(task1Checkpoints))
+ .anyTimes();
+ EasyMock.expect(indexTaskClient.getStatusAsync(EasyMock.anyString()))
+ .andReturn(Futures.immediateFuture(SeekableStreamIndexTaskRunner.Status.READING))
+ .anyTimes();
+ EasyMock.expect(indexTaskClient.getStartTimeAsync(EasyMock.anyString()))
+ .andReturn(Futures.immediateFuture(DateTimes.nowUtc()))
+ .anyTimes();
+ EasyMock.expect(indexTaskClient.getCurrentOffsetsAsync(EasyMock.anyString(), EasyMock.anyBoolean()))
+ .andReturn(Futures.immediateFuture(ImmutableMap.of("0", "100")))
+ .anyTimes();
+ EasyMock.expect(taskRunner.getRunningTasks()).andReturn(ImmutableList.of()).anyTimes();
+
+ final Capture<Task> submittedTasks = Capture.newInstance(CaptureType.ALL);
+ EasyMock.expect(taskQueue.add(EasyMock.capture(submittedTasks))).andReturn(true).anyTimes();
+
+ replayAll();
+
+ final TestSeekableStreamSupervisor supervisor = new TestSeekableStreamSupervisor();
+ supervisor.start();
+ // Pre-populate an actively-reading task group with the single running task. This keeps isBoundedWorkComplete()
+ // false (there is an active group), so runInternal() flows through to createNewTasks() rather than short-circuiting
+ // on completion, and it models the real churn scenario where a completed replica has exited leaving fewer tasks
+ // than the configured replica count.
+ supervisor.addTaskGroupToActivelyReadingTaskGroup(
+ 0,
+ ImmutableMap.of("0", "100"),
+ null,
+ null,
+ Set.of("task1"),
+ Set.of(),
+ Map.of()
+ );
+ supervisor.runInternal();
+
+ Assertions.assertFalse(
+ submittedTasks.hasCaptured(),
+ "No replica top-up task should be submitted for a bounded task group that has reached its end offsets"
+ );
+ Assertions.assertEquals(1, supervisor.getActiveTaskGroup(0).tasks.size());
+
+ verifyAll();
+ }
+
/**
* Verifies that when N stored offsets are all outside stream retention, all N partitions are collected in a single
* pass and reset together in one {@code resetInternal()} call, producing exactly one exception event.
@@ -3746,6 +3862,14 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
super(partitions);
}
+ public TestSeekableStreamDataSourceMetadata(
+ SeekableStreamSequenceNumbers<String, String> partitions,
+ BoundedStreamConfig boundedStreamConfig
+ )
+ {
+ super(partitions, boundedStreamConfig);
+ }
+
@Override
public DataSourceMetadata asStartMetadata()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org