showuon commented on code in PR #12603:
URL: https://github.com/apache/kafka/pull/12603#discussion_r967027203
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -756,10 +756,15 @@ protected boolean onJoinPrepare(Timer timer, int
generation, String memberId) {
joinPrepareTimer.update();
}
+ final SortedSet<TopicPartition> partitionsToRevoke =
getPartitionsToRevoke(protocol, generation, memberId);
+
// async commit offsets prior to rebalance if auto-commit enabled
// and there is no in-flight offset commit request
- if (autoCommitEnabled && autoCommitOffsetRequestFuture == null) {
- autoCommitOffsetRequestFuture = maybeAutoCommitOffsetsAsync();
+ if (autoCommitEnabled) {
+ pausePartitions(partitionsToRevoke);
Review Comment:
I think what Jason suggested, is change from "committing offset subscribed
partitions -> revoking", to "revoking -> "committing offset subscribed
partitions(including revoked partitions)". I don't think it is a good idea
because that would open a door to allow consumer committing offsets for
partitions it doesn't own.
--
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
