[
https://issues.apache.org/jira/browse/KAFKA-20724?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090685#comment-18090685
]
Eswarar Siva commented on KAFKA-20724:
--------------------------------------
Ha, agreed, this one needed the planets to line up :) a rarely used feature
plus a window two map operations wide. Not something you trip over on a quiet
afternoon...but Good that it surfaced before someone met it in production though
On impact, agreed it should be uncommon in practice. Still worth closing, since
when it does fire it takes the whole client down.
I would be happy to take a shot at the PR. I will go with the lock based
direction you put first, holding a lock across the two map writes in pauseTask
and resumeTask so tasks() never sees the task in both, consistent with how the
restored and failed task transitions are already protected. I will add a unit
test that drives the duplicate through tasks(), run it against my reproduction
to confirm both the ISE and the Duplicate key crash are gone, and open it
against trunk linked here.
nd for the separate crash: agreed the transitToUpdateStandby throw is an
invariant check, not the root cause. I think this one is independent of the
duplicate listing, though. resumeTask runs only on the state updater thread and
calls transitToUpdateStandby, which its own comment notes is not idempotent,
guarded only by updatingTasks.size() == 1 and not by whether the reader is
still ACTIVE_RESTORING. You can reach it on the updater thread alone, with no
duplicate involved: when the reader is already in STANDBY_UPDATING, pausing the
updating tasks leaves the reader as is and then resuming a single standby takes
the size == 1 branch and calls transitToUpdateStandby again. So a lock around
the pause and resume map transition would not cover that path. It may still be
in scope for your fix, so I will check it against my repro once the fix is in
and raise it separately with the detail if it survives.
> Topology pausing may have race condition
> ----------------------------------------
>
> Key: KAFKA-20724
> URL: https://issues.apache.org/jira/browse/KAFKA-20724
> Project: Kafka
> Issue Type: Task
> Components: streams
> Affects Versions: 4.0.2, 4.1.2, 4.2.1, 4.4.0
> Reporter: Lucas Brutschy
> Priority: Minor
>
> h2. Summary
> {{DefaultStateUpdater.tasks()}} can return the same task twice when a
> topology pause or resume is in progress, causing {{handleAssignment}} and
> {{handleLostAll}} to queue two REMOVE actions for the same task ID. The
> second REMOVE finds nothing and completes its future with null.
> {{TaskManager.waitForFuture}} treats null as a fatal bug and throws
> {{{}IllegalStateException{}}}, killing the StreamThread.
> h2. Root cause
> {{pauseTask}} and {{resumeTask}} in {{StateUpdaterThread}} move tasks between
> {{updatingTasks}} and {{pausedTasks}} with no lock held:
>
> {{// pauseTask — no lock
> pausedTasks.put(taskId, task); // task now visible in pausedTasks
> updatingTasks.remove(taskId); // task still visible in updatingTasks until
> here}}
>
> {{// resumeTask — no lock
> updatingTasks.put(taskId, task); // task now visible in updatingTasks
> pausedTasks.remove(taskId); // task still visible in pausedTasks until
> here}}
> {{DefaultStateUpdater.tasks()}} calls {{{}executeWithQueuesLocked{}}}, which
> holds {{{}tasksAndActionsLock{}}}, {{{}restoredActiveTasksLock{}}}, and
> {{exceptionsAndFailedTasksLock}} — but no lock covering {{updatingTasks}} or
> {{{}pausedTasks{}}}. {{streamOfTasks()}} streams {{updatingTasks}} and
> {{pausedTasks}} as two separate sub-streams. If either of the above
> transitions is in flight, the task appears in both, and {{tasks()}} returns
> two distinct {{ReadOnlyTask}} wrappers (no {{{}equals{}}}/{{{}hashCode{}}}
> override on {{{}ReadOnlyTask{}}}).
> h2. Why that becomes fatal
> {{handleTasksInStateUpdater}} (called from {{{}handleAssignment{}}}) and
> {{removeLostActiveTasksFromStateUpdaterAndPendingTasksToInit}} (called from
> {{{}handleLostAll{}}}) iterate {{stateUpdater.tasks()}} and queue a REMOVE
> per task into a {{{}LinkedHashMap<TaskId, CompletableFuture>{}}}. A duplicate
> task ID means {{put}} silently overwrites the first future. Two REMOVE
> actions are now in the queue but only the second future is tracked.
> The StateUpdater thread processes both REMOVEs. The first finds the task and
> removes it (completes the orphaned first future that nobody waits on). The
> second finds nothing and calls {{{}future.complete(null){}}}. Back on the
> StreamThread, {{waitForFuture}} sees null and throws:
>
> {{IllegalStateException: Task X was not found in the state updater. This
> indicates a bug.}}
> This propagates out of the rebalance callback, the default uncaught exception
> handler shuts down the client.
> h2. Affected paths
> * {{handleAssignment}} → {{handleRestoringAndUpdatingTasks}} →
> {{handleTasksInStateUpdater}} (line 606): triggered on every assignment when
> topology is paused/resuming.
> * {{handleLostAll}} →
> {{removeLostActiveTasksFromStateUpdaterAndPendingTasksToInit}} (line 1230):
> triggered on producer fence with paused topology.
> {{revokeTasksInStateUpdater}} is NOT affected: it removes matched partitions
> from {{remainingRevokedPartitions}} after each task, so the second occurrence
> of a duplicate task fails the {{containsAll}} guard and no second REMOVE is
> queued.
> h2. Fix
> Make {{pauseTask}} and {{resumeTask}} hold {{restoredActiveTasksLock}} (or a
> dedicated lock) during the two-step transition, consistent with how
> {{addToRestoredTasks}} and
> {{addToExceptionsAndFailedTasksThenRemoveFromUpdatingTasks}} protect their
> equivalent transitions. Alternatively, add a guard in
> {{handleTasksInStateUpdater}} (and the lost-all path) analogous to the
> {{remainingRevokedPartitions}} guard in {{revokeTasksInStateUpdater}} — skip
> a task ID already present in the futures map.
> h2. Reproduction condition
> Any topology that calls {{KafkaStreams.pauseTopology()}} or
> {{resumeTopology()}} during a rebalance. The window is two
> {{ConcurrentHashMap}} operations wide, so it requires timing luck but no
> injected delay.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
