[GitHub] [kafka] guozhangwang commented on a diff in pull request #12439: KAFKA-10199: Further refactor task lifecycle management

[GitBox <gi...@apache.org>]

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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org