[GitHub] [kafka] guozhangwang commented on a change in pull request #11911: KAFKA-13463: Make pause behavior consistent between cooperative and eager protocols

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

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



##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##########
@@ -456,6 +457,12 @@ protected void onJoinComplete(int generation,
 
         subscriptions.assignFromSubscribed(assignedPartitions);
 
+        if (pausedPartitions != null && !pausedPartitions.isEmpty())

Review comment:
       Why not always try to get from `subscriptions.pausedPartitions()`, than maintaining a local copy? The paused partitions could change at any time in between as the onJoinXXX may be triggered by the hb thread right?

##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##########
@@ -762,6 +769,7 @@ else if (future.failed() && !future.isRetriable()) {
         // so that users can still access the previously owned partitions to commit offsets etc.
         Exception exception = null;
         final SortedSet<TopicPartition> revokedPartitions = new TreeSet<>(COMPARATOR);
+        pausedPartitions = subscriptions.pausedPartitions();

Review comment:
       I'm wondering if we should also clear the fetched messages in the buffer when revoking the partitions as well? As suggested in the ticket itself:
   
   ```
   corresponding message in the memory (Fetcher.completedFetches) of pausedPartitions was not cleared, resulting in Fetcher#fetchedRecords() still fetching the message
   ```




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