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

mjsax · web-flow · commit aa016f1bec81 · 2025-12-19T16:15:58.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
@@ -1793,9 +1793,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
`@@ -1793,9 +1793,7 @@ void maybeInitTaskTimeoutsOrThrow(`
`1793`	`1793`	`final long nowMs`
`1794`	`1794`	`) {`
`1795`	`1795`	`for (final TopicPartition partition : partitions) {`
`1796`		`- final Task task = getActiveTask(partition);`
`1797`		`- task.maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);`
`1798`		`- stateUpdater.add(task);`
	`1796`	`+ getActiveTask(partition).maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);`
`1799`	`1797`	`}`
`1800`	`1798`	`}`
`1801`	`1799`