This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new f4c3454b220 KAFKA-20827: don't re-create an already-owned task on
assignment (#22925)
f4c3454b220 is described below
commit f4c3454b220545c4a17908f7583a765c95cb606d
Author: Alan Lau <[email protected]>
AuthorDate: Fri Jul 24 18:29:39 2026 -0400
KAFKA-20827: don't re-create an already-owned task on assignment (#22925)
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.
Reviewers: Bill Bejeck <[email protected]>
---
.../streams/processor/internals/TaskManager.java | 6 +++
.../processor/internals/TaskManagerTest.java | 45 ++++++++++++++++++++++
2 files changed, 51 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 82f47c37abb..24d06638691 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
@@ -462,6 +462,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<StreamTask> newActiveTasks =
activeTaskCreator.createTasks(mainConsumer, activeTasksToCreate);
final Collection<StandbyTask> 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 553a84d4a37..44e8eb05082 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
@@ -291,6 +291,51 @@ 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.
+ 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".
+ 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)
