This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new 895dc49560f KAFKA-20827: don't re-create an already-owned task on
assignment (4.2) (#22933)
895dc49560f is described below
commit 895dc49560f5b41dc44ce8ddbbba7eabc2ae8b33
Author: Alan Lau <[email protected]>
AuthorDate: Tue Aug 4 08:54:23 2026 -0400
KAFKA-20827: don't re-create an already-owned task on assignment (4.2)
(#22933)
Issue: https://issues.apache.org/jira/browse/KAFKA-20827
Backport of #22925 to 4.2, adapted for the older branch:
- createNewTasks keeps its Collection<Task> locals; trunk narrowed
ActiveTaskCreator/StandbyTaskCreator.createTasks to
Collection<StreamTask>/Collection<StandbyTask>, but on 4.2 they still
return Collection<Task>.
- The regression tests build the TaskManager with
setUpTaskManagerWithStateUpdater(...); 4.2's default test setup uses a
null state updater that checkStateUpdater would dereference
A task that fails to (re)initialize is left owned (registered and
flagged failed) to be reconciled via the corruption/failed-task path. It
is excluded from handleAssignment's rectify-existing pass (which
iterates only non-failed tasks), so the next assignment builds a SECOND
representation of it in createNewTasks; that duplicate fails init and
trips the single-owner invariant in Tasks.addStandbyTask/addActiveTask
(IllegalStateException: 'Attempted to create an standby task that we
already own'), killing the StreamThread from an otherwise recoverable
path.
createNewTasks now skips creating any task the registry already owns
(including failed ones), so the duplicate is never built and the
single-owner invariant stays strict. Adds a TaskManagerTest regression
test.
[Systems Tests
Passing](https://confluent-open-source-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/trunk/2026-07-27--001.2d2524ae-1188-4a8f-acb4-6f03b9a5bc6d--1785187224--alanlau28--4.2-KAFKA-20827--0e15adbd26/report.html)
Reveiwers: Bill Bejeck <[email protected]>
---
.../streams/processor/internals/TaskManager.java | 6 +++
.../processor/internals/TaskManagerTest.java | 49 ++++++++++++++++++++++
2 files changed, 55 insertions(+)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
index 6486e9e6e99..4e5947a9df4 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
@@ -448,6 +448,12 @@ public class TaskManager {
private void createNewTasks(final Map<TaskId, Set<TopicPartition>>
activeTasksToCreate,
final Map<TaskId, Set<TopicPartition>>
standbyTasksToCreate) {
+ // A task that failed to initialize is left owned (registered and
flagged failed) for the corruption/
+ // failed-task path to reconcile, and the rectify pass above skips
failed tasks -- so without this guard
+ // the assignment would build a second representation here and trip
the single-owner invariant in Tasks.
+ activeTasksToCreate.keySet().removeIf(tasks::containsInitialized);
+ standbyTasksToCreate.keySet().removeIf(tasks::containsInitialized);
+
final Collection<Task> newActiveTasks =
activeTaskCreator.createTasks(mainConsumer, activeTasksToCreate);
final Collection<Task> newStandbyTasks =
standbyTaskCreator.createTasks(standbyTasksToCreate);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
index 579b5e1fc6a..b4ba69e01c8 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
@@ -319,6 +319,55 @@ public class TaskManagerTest {
verify(schedulingTaskManager).unlockTasks(Set.of(taskId00, taskId01));
}
+ @Test
+ public void shouldNotRecreateActiveTasksThatAreAlreadyOwned() {
+ // Real-registry regression test reproducing the crash end-to-end on
the active path (reached with
+ // num.standby.replicas=0). A task whose init throws is left
owned-and-failed; handleAssignment's rectify
+ // pass skips failed tasks, so a second assignment used to build a
second representation in createNewTasks
+ // that, once it also failed init, tripped the single-owner invariant
in Tasks and killed the StreamThread.
+ // checkStateUpdater drives the pending-init drain, so build with a
state updater (the shared field has none).
+ final TaskManager taskManager =
setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, null);
+ final StreamTask task00 = statefulTask(taskId00,
taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.CREATED).build();
+ when(activeTaskCreator.createTasks(consumer,
taskId00Assignment)).thenReturn(singletonList(task00));
+ doThrow(new
RuntimeException("KABOOM!")).when(task00).initializeIfNeeded();
+
+ // first assignment: 0_0 is created, its init fails, and it is left
owned-but-failed in the registry
+ taskManager.handleAssignment(taskId00Assignment, emptyMap());
+ assertThrows(StreamsException.class, () ->
taskManager.checkStateUpdater(time.milliseconds(), noOpResetter));
+
+ // second assignment of the same owned-but-failed task must not build
a second representation
+ taskManager.handleAssignment(taskId00Assignment, emptyMap());
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter); //
must not throw "already own: 0_0"
+
+ verify(activeTaskCreator, times(1)).createTasks(consumer,
taskId00Assignment);
+ }
+
+ @Test
+ public void shouldNotRecreateStandbyTasksThatAreAlreadyOwned() {
+ // The same guard must protect the standby side -- the original
standby/active recycle crash, reached with
+ // num.standby.replicas>=1. An owned-but-failed standby that is
re-assigned must not be rebuilt, or the
+ // duplicate trips "Attempted to create an standby task that we
already own".
+ // checkStateUpdater drives the pending-init drain, so build with a
state updater (the shared field has none).
+ final TaskManager taskManager =
setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, null);
+ final StandbyTask task00 = standbyTask(taskId00,
taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.CREATED).build();
+
when(standbyTaskCreator.createTasks(taskId00Assignment)).thenReturn(singletonList(task00));
+ doThrow(new
RuntimeException("KABOOM!")).when(task00).initializeIfNeeded();
+
+ // first assignment: 0_0 standby is created, its init fails, and it is
left owned-but-failed
+ taskManager.handleAssignment(emptyMap(), taskId00Assignment);
+ assertThrows(StreamsException.class, () ->
taskManager.checkStateUpdater(time.milliseconds(), noOpResetter));
+
+ // second assignment of the same owned-but-failed standby must not
build a second representation
+ taskManager.handleAssignment(emptyMap(), taskId00Assignment);
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter); //
must not throw "already own" (standby)
+
+ verify(standbyTaskCreator, times(1)).createTasks(taskId00Assignment);
+ }
+
@Test
public void shouldLockAffectedTasksOnHandleRevocation() {
final StreamTask activeTask1 = statefulTask(taskId00,
taskId00ChangelogPartitions)
