zoro30102000 opened a new pull request, #22649: URL: https://github.com/apache/kafka/pull/22649
JIRA: https://issues.apache.org/jira/browse/KAFKA-20724 DefaultStateUpdater.tasks() can return the same TaskId twice. The snapshot is taken under executeWithQueuesLocked, which holds tasksAndActionsLock, restoredActiveTasksLock and exceptionsAndFailedTasksLock, but not updatingTasks or pausedTasks. pauseTask does pausedTasks.put(id) then updatingTasks.remove(id), and resumeTask does the reverse, so for a short window the task sits in both maps. streamOfTasks() reads both, and ReadOnlyTask has no equals/hashCode, so the returned Set keeps both wrappers. A caller that does one remove per entry then removes the same task twice. The first remove succeeds; the second reaches removeTask, finds the task in none of the collections, and completes the future with null, so waitForFuture turns it into "Task X was not found in the state updater. This indicates a bug." The same duplicate also breaks TaskManager.allTasks(), which uses Collectors.toMap, with "Duplicate key". Fix: take restoredActiveTasksLock (the same lock tasks() already holds) around the two map writes in pauseTask and resumeTask, so a reader never observes the task in both maps. The changelogReader transitions stay outside the lock. Tests: two tests in DefaultStateUpdaterTest hold restoredActiveTasksLock and assert that the pause and resume transitions block on it. They fail by timeout if the lock is dropped from either method. restoredActiveTasksLock was relaxed to package private so the tests can take it. Noticed while here, not fixed in this PR: the standalone updatingTasks() and pausedTasks() accessors snapshot without restoredActiveTasksLock, so a caller that composes both itself could still see an inconsistent pair. There is no such caller today (pausedTasks() is test only), so I left it out. Happy to fold it in here or track it separately. ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
