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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
