[
https://issues.apache.org/jira/browse/KAFKA-20724?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090669#comment-18090669
]
Eswarar Siva commented on KAFKA-20724:
--------------------------------------
this matches what I was chasing. I have a deterministic reproduction of exactly
this duplicate listing.
It is a standalone Streams app, several KafkaStreams instances in one JVM,
exactly_once_v2, against a real broker, with cold start churn plus
KafkaStreams.pause() and resume() (the global pause, which drives the same
pauseTask and resumeTask transition you describe). The window is normally sub
microsecond, so to make it deterministic I add a Byteman rule that sleeps right
after the put in pauseTask and resumeTask. That only widens the existing gap
between the put and the remove, no logic change. With it, the ISE through
addToTasksToClose and the Duplicate key crash through allTasks() both fire
within a couple of minutes. With pause and resume turned off I get no
duplicates and no crash at all, even under heavy kill all and restart all
churn, which matches the window being in pauseTask and resumeTask.
One data point for the affects list: I reproduced it on 4.1.2 and on 4.3.0. The
ticket lists 4.0.2, 4.1.2, 4.2.1 and 4.4.0, so 4.3.0 looks affected too. I can
attach the reproducer, the Byteman rule and DEBUG logs (TaskManager,
DefaultStateUpdater and StoreChangelogReader) if useful.
Under the same churn I also hit a separate crash that looks like a different
bug in the same area: resumeTask calls changelogReader.transitToUpdateStandby()
with no guard on the reader state, and that throws when the reader is not
ACTIVE_RESTORING, killing the StreamThread with "The changelog reader is not
restoring active tasks (is STANDBY_UPDATING)". That looks like the KAFKA-17946
area rather than this ticket. Happy to raise it wherever you prefer.
> 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)
