[jira] [Comment Edited] (KAFKA-20721) Streams: "Task not found in the state updater. This indicates a bug." after a state updater remove times out during cold start rebalancing

["Eswarar Siva (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-20721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090890#comment-18090890 ] 

Eswarar Siva edited comment on KAFKA-20721 at 6/23/26 11:35 AM:
----------------------------------------------------------------

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


was (Author: JIRAUSER313660):
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)