This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new 3870a7aca2d KAFKA-20827: don't re-create an already-owned task on
assignment (#22925)
3870a7aca2d is described below
commit 3870a7aca2d74c8f27e3ecd1afdf92395d7a9730
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 1657b3a414b..655598e692f 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
@@ -455,6 +455,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 cc3cd12b918..b1fb17c64b9 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
@@ -289,6 +289,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)
