This is an automated email from the ASF dual-hosted git repository.
FrankChen021 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 587e781a1bd fix: Do not release a stopped streaming task to publish
after a prior task group times out (#20651)
587e781a1bd is described below
commit 587e781a1bd72066711553b4da6ad696251aba0c
Author: Andreas Maechler <[email protected]>
AuthorDate: Wed Oct 21 02:58:06 2026 -0600
fix: Do not release a stopped streaming task to publish after a prior task
group times out (#20651)
* Do not release a stopped streaming task to publish after a prior group
times out
When a pending-completion task group hits its completion timeout, the
supervisor kills it and stops every later group on the same partitions.
It then completes the timed-out group, which releases the publish wait of
a later task that is being stopped. The Overlord commits that task's
segments and offsets while the task is interrupted, so the task never sees
the publish succeed and does not emit ingest/rows/published.
Complete the publish wait of a task with "not ready" whenever the
supervisor stops or kills it.
Fixes #20604
* Release the pending publish wait in killTaskWithSuccess too
---
.../supervisor/SeekableStreamSupervisor.java | 20 +++++++++--
.../SeekableStreamSupervisorStateTest.java | 41 ++++++++++++++++++++++
2 files changed, 59 insertions(+), 2 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..0d532025a2a 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
@@ -1074,8 +1074,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
/**
* Tasks waiting on other task groups to finish publishing to overlapping
partitions.
- * Map from taskId to {@link TaskWaitingToPublish}. This map is maintained
only
- * for reporting purposes.
+ * Map from taskId to {@link TaskWaitingToPublish}.
*/
private final ConcurrentHashMap<String, TaskWaitingToPublish>
tasksWaitingToPublish = new ConcurrentHashMap<>();
@@ -2405,6 +2404,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
StringUtils.format(reasonFormat, args)
);
} else {
+ unblockPublishAsNotReady(id);
taskQueue.get().shutdown(id, reasonFormat, args);
}
} else {
@@ -2412,6 +2412,18 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
}
+ /**
+ * Completes the pending publish wait of the given task, if any, with a "not
ready"
+ * response so that a task which is being stopped or killed does not commit
its offsets.
+ */
+ private void unblockPublishAsNotReady(String taskId)
+ {
+ final TaskWaitingToPublish waitingTask =
tasksWaitingToPublish.remove(taskId);
+ if (waitingTask != null) {
+ waitingTask.readyToPublish.set(false);
+ }
+ }
+
private void killTaskWithSuccess(final String id, String reasonFormat,
Object... args)
{
Optional<TaskQueue> taskQueue = taskMaster.getTaskQueue();
@@ -2424,6 +2436,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
StringUtils.format(reasonFormat, args)
);
} else {
+ unblockPublishAsNotReady(id);
taskQueue.get().shutdownWithSuccess(id, reasonFormat, args);
}
} else {
@@ -3205,6 +3218,9 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
private ListenableFuture<Void> stopTask(final String id, final boolean
publish)
{
+ if (!publish) {
+ unblockPublishAsNotReady(id);
+ }
return Futures.transform(
taskClient.stopAsync(id, publish), new Function<>()
{
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..e265ab9c5df 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
@@ -591,6 +591,47 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
Assertions.assertFalse(laterTaskGroup.completed.get());
}
+ @Test
+ public void
test_isTaskReadyToPublishSegments_returnsFalse_ifTaskIsStoppedAfterEarlierTaskTimesOut()
+ {
+ EasyMock.expect(spec.isSuspended()).andReturn(false).anyTimes();
+
EasyMock.expect(recordSupplier.getPartitionIds(STREAM)).andReturn(ImmutableSet.of(SHARD_ID)).anyTimes();
+
EasyMock.expect(taskQueue.getActiveTasksForDatasource(DATASOURCE)).andReturn(Map.of()).anyTimes();
+
EasyMock.expect(taskQueue.add(EasyMock.anyObject())).andReturn(true).anyTimes();
+ EasyMock.expect(taskStorage.getStatus("earlierTask"))
+
.andReturn(Optional.of(TaskStatus.running("earlierTask"))).anyTimes();
+ EasyMock.expect(taskStorage.getStatus("laterTask"))
+
.andReturn(Optional.of(TaskStatus.running("laterTask"))).anyTimes();
+
+ taskQueue.shutdown(EasyMock.eq("earlierTask"), EasyMock.anyString(),
EasyMock.anyObject());
+ EasyMock.expectLastCall().atLeastOnce();
+ EasyMock.expect(indexTaskClient.stopAsync("laterTask", false))
+ .andReturn(Futures.immediateFuture(true)).atLeastOnce();
+
+ replayAll();
+
+ final TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor();
+ supervisor.start();
+ supervisor.addDiscoveredTaskToPendingCompletionTaskGroups(0,
"earlierTask", Map.of(SHARD_ID, "0"));
+ supervisor.addDiscoveredTaskToPendingCompletionTaskGroups(0, "laterTask",
Map.of(SHARD_ID, "100"));
+
+ final ListenableFuture<Boolean> future =
supervisor.isTaskReadyToPublishSegments(
+ "laterTask",
+ new TestSeekableStreamDataSourceMetadata(
+ new SeekableStreamStartSequenceNumbers<>(STREAM, Map.of(SHARD_ID,
"100"), Set.of())
+ )
+ );
+ Assertions.assertFalse(future.isDone());
+
+ // Expire the earlier task group so that the supervisor kills it and stops
the later task group
+ supervisor.getPendingCompletionTaskGroups(0).get(0).completionTimeout =
DateTimes.nowUtc().minusMinutes(1);
+ supervisor.runInternal();
+
+ Assertions.assertTrue(future.isDone());
+ Assertions.assertEquals(Boolean.FALSE, FutureUtils.getUnchecked(future,
true));
+ verifyAll();
+ }
+
@Test
public void testConnectingToStreamFail()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
