[
https://issues.apache.org/jira/browse/KAFKA-20721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090890#comment-18090890
]
Eswarar Siva commented on KAFKA-20721:
--------------------------------------
Agreed and I think one more piece has to go with it.
That timeout comes from KAFKA-19831 (#20767) right. It was added so
shutdownStateUpdater would not hang when the state updater thread is dead and
the hang it avoids is true: when the state updater thread exits, run() goes
through its finally and calls clearInputQueue(), which clears
tasksAndActions without completing the pending removal futures. Any remove()
still queued at that point never completes, so a plain future.get() blocks
forever. #20767 covered that with future.get(5, TimeUnit.MINUTES).
The problem is the recovery it chose. On TimeoutException waitForFuture logs
the warning and returns null, and both callers, getNonFailedTasks and
addToTasksToClose, filter the nulls out. So the task is dropped on the
StreamThread side while its remove is still queued in the updater.
On the next assignment the task is still in stateUpdater.tasks(), a second
remove is enqueued, that one finds the task in none of the collections and
completes the future with null and waitForFuture turns that null into "was not
found ... This indicates a bug". That is the
production ordering exactly: timeout warning, then "could not be removed ...
does not own this task", then the ISE.
So I do not think the timeout can just be dropped on its own or we are back to
the original hang on a dead updater. What I would suggest:
1. Make the state updater complete its outstanding removal futures when the
thread stops. In run()'s finally, before clearing tasksAndActions, complete
every pending REMOVE future exceptionally with the failure that stopped the
thread. After that a dead updater no longer leaves a future hanging, which
removes the reason the timeout existed.
2. Stop returning null from waitForFuture on timeout. A removal then has three
outcomes: a real result, an exceptional completion that we route to a dirty
close, or future.complete(null) from removeTask, which is the genuine "task is
not here" case and should keep throwing the ISE so we still catch the
pause/resume duplicate from KAFKA-20724.
3. If we still want a backstop for a state updater that is alive but stuck,
bound it to the rebalance timeout instead of a fixed 5 minutes and treat it
fatally, as you said. A removal that cannot finish within the rebalance timeout
means we are out of the group anyway, so failing the thread cleanly and letting
it rejoin is correct and never leaves a half removed task behind. The one
wrinkle is shutdownStateUpdater, which already force closes whatever is left
after its own timeout, so the fatal behavior should sit on the live revoke and
assignment paths, not on shutdown.
Net effectwill be => a removal is always either completed or fails the thread
cleanly, never silently skipped. I am happy to put up a PR along these lines.
One question for the backstop in 3: should the rebalance timeout be read from
the configured value, since TaskManager does not hold it tod
> Streams: "Task not found in the state updater. This indicates a bug." after a
> state updater remove times out during cold start rebalancing
> ------------------------------------------------------------------------------------------------------------------------------------------
>
> 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)
