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

mjsax · mjsax · commit 3c62b6b87442 · 2025-12-19T16:29:50.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
@@ -1856,9 +1856,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
`@@ -1856,9 +1856,7 @@ void maybeInitTaskTimeoutsOrThrow(`
`1856`	`1856`	`final long nowMs`
`1857`	`1857`	`) {`
`1858`	`1858`	`for (final TopicPartition partition : partitions) {`
`1859`		`- final Task task = getActiveTask(partition);`
`1860`		`- task.maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);`
`1861`		`- stateUpdater.add(task);`
	`1859`	`+ getActiveTask(partition).maybeInitTaskTimeoutOrThrow(nowMs, timeoutException);`
`1862`	`1860`	`}`
`1863`	`1861`	`}`
`1864`	`1862`