[
https://issues.apache.org/jira/browse/KAFKA-20724?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090894#comment-18090894
]
Eswarar Siva commented on KAFKA-20724:
--------------------------------------
[~lucasbru] Will do. I will open the PR for this one (lock the pauseTask and
resumeTask map transitions so tasks() can never list the same id twice, with
tests that the duplicate listing, the double remove ISE and the Duplicate key
crash all go away).
transitToUpdateStandby=> agreed on the invariant and I will file a separate
bug and link it here (https://issues.apache.org/jira/browse/KAFKA-20731). Three
places call it on the state updater thread without checking the reader state
first: addTask and resumeTask, each when a standby is added or resumed and
updatingTasks.size() == 1, and transitToUpdateStandbysIfOnlyStandbysLeft. Since
it is documented as not idempotent and throws when the reader is not
ACTIVE_RESTORING, I will propose either guarding all three with
isRestoringActive() or making the transition a no op when the reader is already
STANDBY_UPDATING. Details in the new ticket.
> 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)
