KAFKA-10199: Cleanup TaskManager and Task interfaces #12338
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ommit-in-state-updater
guozhangwang
changed the title
guozhangwang
commented
Jul 7, 2022
| @@ -83,7 +83,6 @@ class DefaultStateUpdaterTest { | |||
| private final Time time = new MockTime(1L); | |||
| private final StreamsConfig config = new StreamsConfig(configProps()); | |||
| private final ChangelogReader changelogReader = mock(ChangelogReader.class); | |||
| private final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = topicPartitions -> { }; | |||
There was a problem hiding this comment.
This is a leftover from previous commit, we do not need this anymore.
| // | ||
| // When that occurs we stash these pending tasks until either they are finally clear to be created, | ||
| // or they are revoked from a new assignment. | ||
| private final Map<TaskId, Set<TopicPartition>> pendingActiveTasks = new HashMap<>(); |
There was a problem hiding this comment.
This is mainly for case 1) in the description.
| // Tasks may have been assigned for a NamedTopology that is not yet known by this host. When that occurs we stash | ||
| // these unknown tasks until either the corresponding NamedTopology is added and we can create them at last, or | ||
| // we receive a new assignment and they are revoked from the thread. | ||
| private final Map<TaskId, Set<TopicPartition>> unknownTasksToBeCreated = new HashMap<>(); |
There was a problem hiding this comment.
This is for case 2) in description.
| @@ -182,7 +159,7 @@ public Collection<Task> createTasks(final Consumer<byte[], byte[]> consumer, | |||
| partitions | |||
| ); | |||
|
|
|||
| final InternalProcessorContext context = new ProcessorContextImpl( | |||
| final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl( | |||
| class Tasks { | ||
| private final Logger log; | ||
| private final TopologyMetadata topologyMetadata; | ||
|
|
||
| private final Map<TaskId, Task> allTasksPerId = Collections.synchronizedSortedMap(new TreeMap<>()); |
There was a problem hiding this comment.
This is a minor cleanup: only return read only tasks upon calling.
| tasksToRecycle, | ||
| tasksToCloseClean, | ||
| tasksToCloseDirty, | ||
| activeTasksToCreate, |
There was a problem hiding this comment.
This is a cleanup: we construct the pending tasks to recycle outside the handleCloseAndRecycle so that we do not need to pass in activeTasksToCreate / standbyTasksToCreate anymore.
| @@ -859,16 +892,6 @@ void closeAndCleanUpTasks(final Collection<Task> activeTasks, final Collection<T | |||
| closeTaskDirty(task); | |||
| } | |||
|
|
|||
| // TODO: change type to `StreamTask` | |||
There was a problem hiding this comment.
This is for case 3) in the description: this is already in the closeTaskClean.
guozhangwang
commented
Jul 7, 2022
| @@ -337,7 +369,7 @@ void addToSuccessfullyProcessed(final Task task) { | |||
| successfullyProcessed.add(task); | |||
| } | |||
|
|
|||
| void removeTaskFromCuccessfullyProcessedBeforeClosing(final Task task) { | |||
| void removeTaskFromSuccessfullyProcessedBeforeClosing(final Task task) { | |||
There was a problem hiding this comment.
Fixed a typo in function name.
3 tasks
|
Closing this PR in favor of #12397 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR is to be reviewed after #12337.
In order to integrate with the state updater, we would need to refactor the TaskManager and Task interfaces. This PR achieved the following purposes:
Separate active and standby tasks in the Tasks placeholder, plus adding
pendingActiveTasksandpendingStandbyTasksinto Tasks. The exposed active/standby tasks from the Tasks set would only be mutated by a single thread, and the pending tasks hold for those tasks that are assigned but cannot be actively managed yet. For now they include two scenarios: a) tasks from unknown sub-topologies and hence cannot be initialized, b) tasks that are pending for being recycled from active to standby and vice versa. Note case b) would be added in a follow-up PR.Extract any logic that mutates a task out of the Tasks / TaskCreators. Tasks should only be a place for maintaining the set of tasks, but not for manipulations of a task; and TaskCreators should only be used for creating the tasks, but not for anything else. These logic are all migrated into TaskManger.
While doing 2) I noticed we have a couple of minor issues in the code where we duplicate the closing logics, so I also cleaned them up in the following way:
a) When closing a task, we first remove it from Tasks, and the trigger the corresponding
closeClean/Dirtyfunction; for active task, we also remove its task producer if EOS-V1 is used.b) For closing dirty, we swallow the exception from
closecall and theremove task producercall; for closing clean, we store the thrown exception from eitherclosecall or theremove task producer, and then rethrow at the end of the caller. The difference though is that, for the exception fromclosecall we need to retry close it dirty; for the exception from theremove task producerwe do not need to re-close it dirty.Committer Checklist (excluded from commit message)