[GitHub] [kafka] guozhangwang commented on a change in pull request #8221: KAFKA-9561: update task input partitions after rebalance

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

guozhangwang commented on a change in pull request #8221:
URL: https://github.com/apache/kafka/pull/8221#discussion_r422583023



##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -429,6 +421,15 @@ public void closeDirty() {
         log.info("Closed dirty");
     }
 
+    @Override
+    public void updateInputPartitions(final Set<TopicPartition> topicPartitions) {
+        super.updateInputPartitions(topicPartitions);
+        partitionGroup.updatePartitions(topicPartitions, recordQueueCreator::createQueue);
+        if (state() != State.RESTORING) { // if task is RESTORING then topology will be initialized in completeRestoration

Review comment:
       If the state was RUNNING then we would effectively call `initializeTopology` twice. Is that intentional?

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
##########
@@ -170,6 +170,8 @@ public boolean isValidTransition(final State newState) {
      */
     void closeDirty();
 
+    void updateInputPartitions(final Set<TopicPartition> topicPartitions);

Review comment:
       These two newly added functions are always called at the same time, I'd suggest we merge them into a single function, e.g. named `update(topicPartitions, topology)`.




----------------------------------------------------------------
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.

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