You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ShivsundarR (via GitHub)" <gi...@apache.org> on 2025/07/28 10:12:43 UTC
[PR] KAFKA-19485 (II) : Complete any pending acknowledgements in ShareFetch on an error response. [kafka]
ShivsundarR opened a new pull request, #20247:
URL: https://github.com/apache/kafka/pull/20247
*What*
Currently when we received a top level error response in ShareFetch, we would log the error, update the share session epoch and proceed to the next request.
But these acknowledgements(if any) are not completed and the callback would not have been processed.
PR aims to address this by completing these acknowledgements with the error code from the response in this case.
--
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-19485 (II) : Complete any pending acknowledgements in ShareFetch on an error response. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #20247:
URL: https://github.com/apache/kafka/pull/20247#discussion_r2246054545
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -758,6 +758,16 @@ private void handleShareFetchSuccess(Node fetchTarget,
if (response.error() == Errors.UNKNOWN_TOPIC_ID) {
metadata.requestUpdate(false);
}
+ // Complete any inFlight acknowledgements with the error code from the response.
+ Map<TopicIdPartition, Acknowledgements> nodeAcknowledgementsInFlight = fetchAcknowledgementsInFlight.get(fetchTarget.id());
+ if (nodeAcknowledgementsInFlight != null) {
+ nodeAcknowledgementsInFlight.forEach((tip, acks) -> {
+ acks.complete(Errors.forCode(response.error().code()).exception());
+ metricsManager.recordFailedAcknowledgements(acks.size());
+ maybeSendShareAcknowledgeCommitCallbackEvent(Map.of(tip, acks));
Review Comment:
Thanks, yeah that makes sense. I have moved it outside the loop now.
--
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-19485 (II) : Complete any pending acknowledgements in ShareFetch on an error response. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20247:
URL: https://github.com/apache/kafka/pull/20247#discussion_r2245606882
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -758,6 +758,16 @@ private void handleShareFetchSuccess(Node fetchTarget,
if (response.error() == Errors.UNKNOWN_TOPIC_ID) {
metadata.requestUpdate(false);
}
+ // Complete any inFlight acknowledgements with the error code from the response.
+ Map<TopicIdPartition, Acknowledgements> nodeAcknowledgementsInFlight = fetchAcknowledgementsInFlight.get(fetchTarget.id());
+ if (nodeAcknowledgementsInFlight != null) {
+ nodeAcknowledgementsInFlight.forEach((tip, acks) -> {
+ acks.complete(Errors.forCode(response.error().code()).exception());
+ metricsManager.recordFailedAcknowledgements(acks.size());
+ maybeSendShareAcknowledgeCommitCallbackEvent(Map.of(tip, acks));
Review Comment:
I think that `nodeAcknowledgementsInFlight` is already the correct type for you just to call `maybeSendShareAcknowledgeCommitCallbackEvent(nodeAcknowledgementsInFlight)` outside the 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19485 (II) : Complete any pending acknowledgements in ShareFetch on an error response. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #20247:
URL: https://github.com/apache/kafka/pull/20247
--
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