AndrewJSchofield commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2290534883
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -859,16 +865,19 @@ public ConsumerRecords<K, V> poll(final Duration timeout)
{
throw new IllegalStateException("Consumer is not subscribed to
any topics or assigned any partitions");
}
- do {
- PollEvent event = new PollEvent(timer.currentTimeMs());
- // Make sure to let the background thread know that we are
still polling.
- // This will trigger async auto-commits of consumed positions
when hitting
- // the interval time or reconciling new assignments
- applicationEventHandler.add(event);
+ PollEvent event = new PollEvent(timer.currentTimeMs());
+ // Make sure to let the background thread know that we are still
polling.
+ // This will trigger async auto-commits of consumed positions when
hitting
+ // the interval time or reconciling new assignments
+ applicationEventHandler.add(event);
+
+ if (reconciliationInProgress.get() ||
autoCommitState.shouldAutoCommit()) {
Review Comment:
I agree with @lianetm that this is opening up the risk of race conditions.
However, I think the principle here is a good one. The risk part here is
related to the auto-commit timer. If auto-commit is not enabled, we absolutely
know that we are not racing with the auto-commit timer. If it is enabled, we
are potentially in a race. So, a slight twist on this can safely optimise when
auto-commit is not enabled.
--
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
