You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "AndrewJSchofield (via GitHub)" <gi...@apache.org> on 2025/10/23 07:49:56 UTC
[PR] KAFKA-19827: Call ack commit callback at end of waiting calls [kafka]
AndrewJSchofield opened a new pull request, #20758:
URL: https://github.com/apache/kafka/pull/20758
The acknowledgement commit callback in the share consumer gets called on the application thread at the start of the poll, commitSync and commitAsync methods. Specifically in the peculiar case of using the callback together with commitSync, the acknowledgement callback for the committed records is called at the start of the next eligible call, even though the information is already known at the end of the commitSync's execution. The results are correct already, but the timing could be improved in some situations.
--
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
Re: [PR] KAFKA-19827: Call ack commit callback at end of waiting calls [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20758:
URL: https://github.com/apache/kafka/pull/20758#discussion_r2483976728
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -609,6 +609,15 @@ public synchronized ConsumerRecords<K, V> poll(final Duration timeout) {
throw e.cause();
} finally {
kafkaShareConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+
+ // Handle any acknowledgements which completed while we were waiting, but do not throw
+ // the exception because the fetched records would then not be returned to the caller
+ try {
+ handleCompletedAcknowledgements(false);
+ } catch (Throwable t) {
+ log.warn("Exception thrown in acknowledgement commit callback", t);
+ }
Review Comment:
@TaiJuWu thanks for this find. The flaky behaviour and its root cause have already been logged in https://issues.apache.org/jira/browse/KAFKA-19840
--
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
Re: [PR] KAFKA-19827: Call ack commit callback at end of waiting calls [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #20758:
URL: https://github.com/apache/kafka/pull/20758#discussion_r2469861779
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -609,6 +609,15 @@ public synchronized ConsumerRecords<K, V> poll(final Duration timeout) {
throw e.cause();
} finally {
kafkaShareConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+
+ // Handle any acknowledgements which completed while we were waiting, but do not throw
+ // the exception because the fetched records would then not be returned to the caller
+ try {
+ handleCompletedAcknowledgements(false);
+ } catch (Throwable t) {
+ log.warn("Exception thrown in acknowledgement commit callback", t);
+ }
Review Comment:
I find this change make `testShareGroupMaxSizeConfigExceeded` become flaky.
The following code will fix this issue but seem to violate the comments.
```
// Handle any acknowledgements which completed while we were waiting, but do not throw
// the exception because the fetched records would then not be returned to the caller
try {
handleCompletedAcknowledgements(false);
} catch (GroupMaxSizeReachedException e) {
log.warn("Exception thrown in acknowledgement commit callback", e);
throw e;
} catch (Throwable t) {
log.warn("Exception thrown in acknowledgement commit callback", t);
}
```
--
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
Re: [PR] KAFKA-19827: Call ack commit callback at end of waiting calls [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #20758:
URL: https://github.com/apache/kafka/pull/20758
--
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
Re: [PR] KAFKA-19827: Call ack commit callback at end of waiting calls [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20758:
URL: https://github.com/apache/kafka/pull/20758#discussion_r2454975033
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -609,6 +609,15 @@ public synchronized ConsumerRecords<K, V> poll(final Duration timeout) {
throw e.cause();
} finally {
kafkaShareConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+
+ // Handle any acknowledgements which completed while we were waiting, but do not throw
+ // the exception because the fetched records would then not be returned to the caller
+ try {
+ handleCompletedAcknowledgements(false);
+ } catch (Throwable t) {
+ log.warn("Failed invoking acknowledgement commit callback", t);
Review Comment:
nit: Please if we can change the log line to say exception happened in vallback but callback was invoked.
--
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