[
https://issues.apache.org/jira/browse/KAFKA-20721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090581#comment-18090581
]
Eswarar Siva commented on KAFKA-20721:
--------------------------------------
[~lucasbru] thanks for pushing back :) ,
Agree , the drain versus revoke mechanism in the description is wrong.
drainRestoredActiveTasks and revokeTasksInStateUpdater both run on the
StreamThread, so they cannot interleave,
and my first tests only showed that calling those two methods from different
threads returns null, which is
expected and not a bug. I withdraw that mechanism.
and here is what I now believe is happening, with a reproduction.
Root cause: DefaultStateUpdater.tasks() can return the same TaskId twice. Two
things combine. First, tasks() takes
its snapshot under executeWithQueuesLocked, which holds tasksAndActionsLock,
restoredActiveTasksLock and
exceptionsAndFailedTasksLock, but not the updatingTasks or pausedTasks maps.
pauseTask does pausedTasks.put(id)
then updatingTasks.remove(id), and resumeTask does the reverse, so for a short
window the task is in both maps and
streamOfTasks() lists it twice. Second, ReadOnlyTask has no equals/hashCode, so
the returned Set does not collapse
the two wrappers.
Once tasks() returns a duplicate, a caller that does one remove per entry
removes the same task twice. The first
remove succeeds, the second reaches removeTask, finds the task in none of the
four collections, and completes the
future with null. waitForFuture then throws "Task X was not found in the state
updater. This indicates a bug." In
my runs it comes through shutdownStateUpdater -> addToTasksToClose. The same
duplicate also breaks
TaskManager.allTasks(), which collects with Collectors.toMap(Task::id, ...) and
throws IllegalStateException:
Duplicate key, killing the StreamThread in runLoop.
On your question about a real application versus unit tests: this is a small
standalone Streams application, several
KafkaStreams instances in one JVM, exactly_once_v2, against a real broker, not
a unit test. The trigger is cold
start churn plus KafkaStreams.pause()/resume(). The duplicate window is
normally sub microsecond, so to make it
deterministic I widened that existing window with a Byteman rule that sleeps
right after the put in pauseTask and
resumeTask (no logic change, it only stretches the gap that is already there).
With that, the ISE and the Duplicate
key error both fire within a couple of minutes. I reproduced it this way on
4.1.2 and 4.3.0; trunk has the same code
on this path. With pause/resume disabled I get no duplicates and no crash at
all, even under aggressive kill all and
restart all churn, which matches the window being in pauseTask/resumeTask.
I attached the DEBUG logs from one run (TaskManager, DefaultStateUpdater and
StoreChangelogReader at DEBUG), the
reproducer (SuBugPoc.java, pom.xml) and the Byteman rule. In the DEBUG log you
can see the same task id present on
the updating and the paused side at the same time, then "could not be removed
from the state updater because the
state updater does not own this task", then the ISE.
On affects version: I reproduced it on 4.1.2 and 4.3.0, so I set those. 4.2.x
and trunk carry the same
pauseTask/resumeTask and ReadOnlyTask code, so they look affected by inspection
as well.
On warn plus skip: I agree with you, I would not just downgrade the ISE. The
throw is doing its job. The fix
probably belongs at the source so tasks() never returns a duplicate, for
example dedupe by task id before wrapping,
or give ReadOnlyTask equals/hashCode based on the task id, or make the
pause/resume transition atomic with respect
to tasks(). I left the ISE as is.
Two related notes from the same digging, in case they help triage:
- KAFKA-17402 is the same family (tasks() returning a duplicate, seen there as
shouldGetTasksFromRestoredActiveTasks
counting 3 instead of 2). Its fix made the restore completion transition
atomic but did not touch
pauseTask/resumeTask, which is the path here.
- My original one off production occurrence was on 4.1.2 with no pause/resume
and very large RocksDB state, which
looks closer to KAFKA-20456 (the state updater stalling and waitForFuture
timing out), already fixed in 4.3.0. So
the production incident and this pause/resume reproduction may be two
different routes to the same exception.
FYI updated the ticket too on the same
> Streams: "Task not found in the state updater. This indicates a bug." caused
> by a duplicate TaskId from DefaultStateUpdater.tasks() under pause/resume
> ------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20721
> URL: https://issues.apache.org/jira/browse/KAFKA-20721
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.3.0, 4.1.2, 4.2.1
> Environment: Kafka Streams, exactly_once_v2, default state updater,
> several stateful tasks. Hit during a cold start with a lot of rebalancing.
> Verified the code at 4.1.2, 4.2.1, 4.3.0 and trunk.
> Reporter: Eswarar Siva
> Assignee: Eswarar Siva
> Priority: Critical
>
> Updated 2026-06-22: the original mechanism in this ticket (a race between
> drainRestoredActiveTasks and
> revokeTasksInStateUpdater) was wrong and is withdrawn. Both run on the
> StreamThread and cannot interleave. The
> corrected root cause and a reproduction are below; see the comments for the
> history.
> A StreamThread dies with:
> java.lang.IllegalStateException: Task X_Y was not found in the state updater.
> This indicates a bug.
> at
> org.apache.kafka.streams.processor.internals.TaskManager.waitForFuture(...)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.addToTasksToClose(...)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.shutdownStateUpdater(...)
> thrown when DefaultStateUpdater.removeTask completes a removal future with
> null.
> Root cause: DefaultStateUpdater.tasks() can return the same TaskId twice.
> tasks() snapshots under
> executeWithQueuesLocked, which holds tasksAndActionsLock,
> restoredActiveTasksLock and exceptionsAndFailedTasksLock,
> but not the updatingTasks or pausedTasks maps. pauseTask does
> pausedTasks.put(id) then updatingTasks.remove(id), and
> resumeTask does the reverse, so for a short window a task is in both maps and
> streamOfTasks() lists it twice.
> ReadOnlyTask has no equals/hashCode, so the returned Set keeps both wrappers.
> A caller that does one remove per
> tasks() entry (shutdownStateUpdater via addToTasksToClose, and
> handleAssignment) then removes the same task twice.
> The first remove succeeds; the second reaches removeTask, finds the task in
> none of the four collections, and
> completes the future with null, so waitForFuture throws the ISE. The same
> duplicate also breaks
> TaskManager.allTasks(), which uses Collectors.toMap(Task::id, ...) and throws
> IllegalStateException: Duplicate key,
> killing the StreamThread in runLoop.
> Reproduction: a standalone Streams application, several KafkaStreams
> instances in one JVM, exactly_once_v2, against a
> real broker, with cold start churn plus KafkaStreams.pause()/resume(). The
> duplicate window is normally sub
> microsecond, so to make it deterministic a Byteman rule sleeps right after
> the put in pauseTask and resumeTask (no
> logic change, it only widens the gap that is already there). With that, the
> ISE and the Duplicate key error both fire
> within a couple of minutes. Reproduced on 4.1.2 and 4.3.0; trunk carries the
> same code on this path. With pause/resume
> disabled there are no duplicates and no crash, even under aggressive churn.
> The reproducer, the Byteman rule and DEBUG
> logs (TaskManager, DefaultStateUpdater, StoreChangelogReader) are attached.
> Affected versions: 4.1.2 and 4.3.0 (reproduced); 4.2.x and trunk by code
> inspection.
> Related: KAFKA-17402 is the same family (a duplicate from tasks(), seen there
> as shouldGetTasksFromRestoredActiveTasks
> counting 3 instead of 2); its fix made the restore completion transition
> atomic but did not touch pauseTask/resumeTask,
> which is the path here. The original production occurrence behind this ticket
> was on 4.1.2 with no pause/resume and very
> large state, which looks closer to KAFKA-20456 (the state updater stalling
> and waitForFuture timing out, fixed in
> 4.3.0), so the production incident and this pause/resume reproduction may be
> two different routes to the same exception.
> Suggested fix: stop tasks() from returning a duplicate, for example dedupe by
> task id before wrapping, or give
> ReadOnlyTask equals/hashCode based on the task id, or make the pause/resume
> transition atomic with respect to tasks().
> The ISE should stay; it is correctly catching the inconsistency.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
