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
