ShivsundarR commented on code in PR #19417:
URL: https://github.com/apache/kafka/pull/19417#discussion_r2044479551
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -673,6 +661,10 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition,
NodeAcknowledgements> ack
// Notify the network thread to wake up and start the next
round of fetching
applicationEventHandler.wakeupNetworkThread();
}
+ if (acknowledgementMode == ShareAcknowledgementMode.EXPLICIT) {
+ // We cannot leave unacknowledged records in EXPLICIT
acknowledgement mode, so we throw an exception to the application.
+ throw new IllegalStateException("There are unacknowledged
records from the previous fetch : " + currentFetch.records());
Review Comment:
Thanks, yes I had removed the records in the last update. I have now updated
the error message as well.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
