philipnee commented on code in PR #12603:
URL: https://github.com/apache/kafka/pull/12603#discussion_r966583573
##########
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:
but if we don't commit these offsets, i wonder if that breaks the poll
contract, i.e. we need to commit the acked data from the previous poll. Note
that, I believe some of the data could already be returned to the client from
the previous poll loop.
--
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
