[
https://issues.apache.org/jira/browse/KAFKA-20721?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Eswarar Siva updated KAFKA-20721:
---------------------------------
Description:
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.
was:
Ran into this on a Streams app running exactly_once_v2, during a cold start
while a bunch of stateful tasks were
still restoring and the group kept rebalancing. One of the StreamThreads just
dies with:
{noformat}
java.lang.IllegalStateException: Task 1_2 was not found in the state updater.
This indicates a bug.
Please report at https://issues.apache.org/jira/projects/KAFKA/issues ...
at
org.apache.kafka.streams.processor.internals.TaskManager.waitForFuture(TaskManager.java:711)
at
org.apache.kafka.streams.processor.internals.TaskManager.lambda$getNonFailedTasks$10(TaskManager.java:672)
at
org.apache.kafka.streams.processor.internals.TaskManager.revokeTasksInStateUpdater(TaskManager.java:1168)
at
org.apache.kafka.streams.processor.internals.TaskManager.handleRevocation(TaskManager.java:1058)
{noformat}
It is thrown inside the rebalance callback, so it comes out as "User rebalance
callback throws an error", the
default handler does SHUTDOWN_CLIENT, and the client goes REBALANCING to
PENDING_ERROR. In my case there was also a
native RocksDB SIGSEGV on the way down. The thread does not recover without a
restart.
After digging into it this is a race, not a real "impossible" state.
revokeTasksInStateUpdater walks stateUpdater.tasks() and calls
stateUpdater.remove(taskId) for every revoked active
task it still sees. tasks() (streamOfNonPausedTasks) also includes tasks
sitting in restoredActiveTasks, meaning
ones that already finished restoring but the stream thread has not drained yet.
If that task leaves every one of the
updater collections (updatingTasks, pausedTasks, restoredActiveTasks,
exceptionsAndFailedTasks) before the state
updater thread actually processes the queued remove (drained by
handleRestoredTasksFromStateUpdater, or pulled out
by another action during the churn), removeTask matches none of the four
branches and does future.complete(null).
waitForFuture then sees the null and throws the fatal "This indicates a bug".
So the task was just already gone by the time the remove ran. That is a benign
lost race and it should not kill the
thread.
It is still there on the latest code. I read the source at 4.1.2, 4.2.1, 4.3.0
and trunk: the throw in waitForFuture
and the future.complete(null) in removeTask are the same, and the
tasks()/streamOfNonPausedTasks that exposes a
restored task is identical between 4.3.0 and trunk. So upgrading does not help.
h3. How to reproduce
I could not get a single thread schedule to fire it reliably (it is a timing
race), so I split it into two
deterministic tests, both green on trunk. Patch attached.
* DefaultStateUpdaterTest: real DefaultStateUpdater, let a task finish
restoring so it lands in restoredActiveTasks,
confirm tasks() still reports it, call drainRestoredActiveTasks(), then
remove(id).get() comes back null.
* TaskManagerTest: drive the real handleRevocation -> revokeTasksInStateUpdater
with tasks() reporting the task and
its remove() future completing null (the outcome the first test shows is real),
and it throws the IllegalStateException
above. The stack trace at the top is the real one from this test.
h3. Fix idea
Treat "task is not in any collection when remove runs" as already removed
instead of a fatal bug. Either have
remove() finish with an explicit already removed result that
revokeTasksInStateUpdater / getNonFailedTasks skip, or
let waitForFuture log and skip on a null result for the revoke path instead of
throwing.
Summary: Streams: "Task not found in the state updater. This indicates
a bug." caused by a duplicate TaskId from DefaultStateUpdater.tasks() under
pause/resume (was: Streams state updater: remove() during rebalance revocation
can complete with null and throw a fatal "Task was not found in the state
updater. This indicates a bug.")
> 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
> Attachments: captured-stacktrace.txt, task-not-found-repro.patch
>
>
> 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)
