KAFKA-20002: Reset-by-duration should not hand back task to state-updater (#21167)

mjsax · mjsax · commit 01b8fca8034b · 2025-12-19T16:23:02.000-08:00
This bug was introduced via KAFKA-18015. Reviewers: Lucas Brutschy <lbrutschy@confluent.io>, Vincent Potuček (@Pankraz76)
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
@@ -1884,9 +1884,7 @@ void maybeInitTaskTimeoutsOrThrow(
         final long nowMs
     ) {
         for (final TopicPartition partition : partitions) {
-            final Task task = getActiveTask(partition);
-            task.maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);
-            stateUpdater.add(task);
+            getActiveTask(partition).maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -1884,9 +1884,7 @@ void maybeInitTaskTimeoutsOrThrow(`
`1884`	`1884`	`final long nowMs`
`1885`	`1885`	`) {`
`1886`	`1886`	`for (final TopicPartition partition : partitions) {`
`1887`		`- final Task task = getActiveTask(partition);`
`1888`		`- task.maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);`
`1889`		`- stateUpdater.add(task);`
	`1887`	`+ getActiveTask(partition).maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);`
`1890`	`1888`	`}`
`1891`	`1889`	`}`
`1892`	`1890`