[
https://issues.apache.org/jira/browse/KAFKA-20721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090663#comment-18090663
]
Eswarar Siva commented on KAFKA-20721:
--------------------------------------
[~lucasbru] Thanks Lucas, thanks for the nudge. Yyes ur right on both counts:
the pause/resume rewrite did not belong on this ticket and the original crash
had no pause/resume. I pulled the production logs for the original occurrence
and it matches KAFKA-20456, so I have reverted the description back to that
original crash.
The app is a Kafka Streams 4.1.2 service on exactly_once_v2, several instances,
that crashed during cold start rebalancing. It never calls pause() or resume().
I found four occurrences and every one follows the same order:
1. A remove times out in the revoke path:
WARN TaskManager: The state updater wasn't able to remove task 1_4 in time. The
state updater thread may be dead.
at TaskManager.waitForFuture(TaskManager.java:709)
at TaskManager.revokeTasksInStateUpdater(TaskManager.java:1220)
at TaskManager.handleRevocation(TaskManager.java:1119)
2. About 13 seconds to 2 minutes later the state updater finally runs a now
duplicate remove for the same task and finds it gone:
WARN DefaultStateUpdater: Task 1_4 could not be removed from the state updater
because the state updater does not
own this task.
3. That null completion is thrown as the ISE, this time from the assignment
close path:
IllegalStateException: Task 1_4 was not found in the state updater. This
indicates a bug.
at TaskManager.waitForFuture(TaskManager.java:711)
at TaskManager.addToTasksToClose(TaskManager.java:685)
at TaskManager.handleAssignment(TaskManager.java:378)
-> User rebalance callback throws an error -> SHUTDOWN_CLIENT
So it is exactly what you described: the 5 minute waitForFuture times out, the
task is left behind, the next rebalance handling issues a second remove for it,
and that second remove completes the future with null, which becomes the ISE.
The timeout WARN comes first in all four cases, so the trigger is a state
updater stall longer than the 5 minute timeout, not pause/resume.
On the trigger itself: it was a cold start with large RocksDB state, and we
have not seen it since moving the app to 4.3.0, which carries the two RocksDB
changes from KAFKA-20456 (the lighter offsets column family and avoiding
persisting closed state). I want to be precise though. I did not capture a
thread dump or GC log at the moment of the stall, so I can say the state
updater stalled past 5 minutes, but I cannot prove the staller was RocksDB
specifically rather than, say, a long GC pause. The "thread may be dead"
message only tells us the get() timed out.
One thing I checked on the fix side: TaskManager.waitForFuture is identical in
4.1.2, 4.3.0 and current trunk (same hardcoded future.get(5, TimeUnit.MINUTES),
same throw on a null completion). The two KAFKA-20456 commits that shipped in
4.3.0 are the RocksDB ones; the timeout bound and the leaked task cleanup are
not in 4.3.0 or trunk. So 4.3.0 makes the stall less likely but the timeout to
duplicate remove to ISE path is still present. The 4.3.0 run has been clean so
far, but the window is short, so I would not read much into that yet.
I set affects version to 4.1.2, where the real occurrence is. Happy to do
whatever is cleanest for tracking: close this as a duplicate of KAFKA-20456, or
keep it open to track that the timeout and duplicate remove protocol is still
unfixed in 4.3.0 and trunk. Your call.
Separately, I have a deterministic pause/resume reproduction for the duplicate
listing crash (Byteman widening the put before remove window in pauseTask and
resumeTask). I will add it to KAFKA-20724, since that is the right home for it.
> Streams: "Task not found in the state updater. This indicates a bug." after a
> state updater remove times out during cold start rebalancing (KAFKA-20456
> family, no 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
> 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
> Attachments: SuBugPoc.java, captured-stacks-4.1.2.txt,
> captured-stacks-4.3.0.txt, poc-debug-excerpt.txt, pom.xml, widen-window.btm
>
>
> Note (2026-06-22): an earlier revision of this ticket described a
> pause/resume duplicate listing mechanism. That was wrong for this report; the
> original production occurrence had no pause/resume. The pause/resume
> duplicate listing is tracked separately in KAFKA-20724. This description is
> reverted to the original crash, which matches KAFKA-20456.
> A StreamThread dies during cold start rebalancing with:
> java.lang.IllegalStateException: Task X_Y was not found in the state updater.
> This indicates a bug.
> at TaskManager.waitForFuture(TaskManager.java:711)
> at TaskManager.addToTasksToClose(TaskManager.java:685)
> at TaskManager.handleTasksInStateUpdater(TaskManager.java:640)
> at TaskManager.handleAssignment(TaskManager.java:378)
> -> KafkaException: User rebalance callback throws an error ->
> SHUTDOWN_CLIENT
> Environment: Kafka Streams 4.1.2, exactly_once_v2, several stateful
> instances, large RocksDB state, cold start with heavy rebalancing. The
> application does not call pause() or resume().
> Mechanism (matches KAFKA-20456): during the rebalance, a remove issued from
> revokeTasksInStateUpdater waits in TaskManager.waitForFuture, which has a
> hardcoded 5 minute timeout. The state updater is stalled (restore under
> load), so the get() times out and waitForFuture returns null after logging
> "The state updater wasn't able to remove task X in time. The state updater
> thread may be dead." The revoked partitions are already removed from
> remainingRevokedPartitions before that wait, so the task is not suspended or
> re tracked. The task is still visible in stateUpdater.tasks() because a
> pending REMOVE action is not included in the snapshot (only pending ADD is).
> The following onAssignment then enumerates tasks(), finds the task absent
> from the new assignment, and enqueues a second remove via
> handleTasksInStateUpdater. When the stalled state updater finally drains its
> action queue (FIFO), the first remove succeeds and the second finds the task
> in none of its collections, so DefaultStateUpdater.removeTask
> completes the future with null and logs "Task X could not be removed from the
> state updater because the state updater does not own this task."
> waitForFuture in addToTasksToClose receives that null and throws the ISE.
> Production evidence: four occurrences on 4.1.2, all the same order, the
> timeout WARN first, then 13 seconds to 2 minutes later the "does not own this
> task" WARN, then the ISE. No STANDBY_UPDATING, no Duplicate key, no pause or
> resume in any of them.
> Relationship to 4.3.0: TaskManager.waitForFuture is identical in 4.1.2, 4.3.0
> and trunk (same 5 minute timeout, same throw on a null completion). The
> KAFKA-20456 changes present in 4.3.0 are the two RocksDB ones (lighter
> offsets column family, avoid persisting closed state), which reduce the
> stalls that trigger the timeout. The timeout bound and the leaked task
> cleanup are not in 4.3.0 or trunk. So 4.3.0 reduces the likelihood but does
> not remove the timeout to duplicate remove to ISE path. No recurrence on
> 4.3.0 so far, but the observation window is short.
> Caveat on the trigger: we did not capture a thread dump or GC log at the
> stall, so we can confirm the state updater stalled past 5 minutes but not
> that the staller was RocksDB specifically rather than, for example, a long GC
> pause.
> Affects version: 4.1.2 (observed). The same code path exists in 4.2.x, 4.3.0
> and trunk.
> Suggested direction: bound the waitForFuture timeout and clean up the leaked
> task when the remove times out, so a slow remove does not leave a task that
> is later removed twice. The ISE should stay; it is correctly catching the
> inconsistency.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
