guozhangwang commented on code in PR #12439:
URL: https://github.com/apache/kafka/pull/12439#discussion_r929297148
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1213,7 +1270,6 @@ void maybeCloseTasksFromRemovedTopologies(final
Set<String> currentNamedTopologi
final Set<Task> allTasksToRemove = union(HashSet::new,
activeTasksToRemove, standbyTasksToRemove);
closeAndCleanUpTasks(activeTasksToRemove, standbyTasksToRemove,
true);
- allTasksToRemove.forEach(tasks::removeTask);
Review Comment:
I found this is redundant code, since in the above `closeAndCleanUpTasks`
we've already triggered `tasks#removeTask`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -77,6 +93,22 @@ public void registerTaskError(final Task task, final
Throwable t, final long now
}
}
+ Collection<Task> successfullyProcessed() {
Review Comment:
This bookkeeping logic is moved from `Tasks`, as part of 2).
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -544,7 +544,7 @@ public void updateInputPartitions(final Set<TopicPartition>
topicPartitions, fin
}
@Override
- public void closeCleanAndRecycleState() {
+ public void recycleAndConvert() {
Review Comment:
We consolidate the `closeCleanAndRecycleState` and the `recycle` below into
this single function now, and extract the creation of the newly recycled task
into the creators, as part of 1). Ditto in StandbyTask.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
