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/04/04 08:16:42 UTC
[PR] KAFKA-19084: Port KAFKA-16224, KAFKA-16764 for ShareConsumers [kafka]
ShivsundarR opened a new pull request, #19369:
URL: https://github.com/apache/kafka/pull/19369
*What*
Currently for ShareConsumers, if we receive an `UNKNOWN_TOPIC_OR_PARTITION` error code in the `ShareAcknowledgeResponse`, then we retry sending the acknowledgements until the timer expires.
We ideally do not want this when a topic/partition is deleted, hence like the `CommitRequestManager`(https://github.com/apache/kafka/pull/15581), we will treat this error as fatal and not retry the acknowledgements.
PR also suppresses `InvalidTopicException` during `unsubscribe()` which was also added in the `AsyncKafkaConsumer`(https://github.com/apache/kafka/pull/16043). It was later removed in the regular consumer here(https://github.com/apache/kafka/commit/d76238a18fb6a86b4b08bc04918ea1a0ee626517), as they notified the background operations of metadata errors instead of propagating them via `ErrorEvent`.
`ShareConsumerImpl` however does not require that change and it still propagates the metadata error back to the application. So we would need to suppress this exception during unsubscribe().
--
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-19084: Port KAFKA-16224, KAFKA-16764 for ShareConsumers [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19369:
URL: https://github.com/apache/kafka/pull/19369#discussion_r2028490277
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -971,9 +971,11 @@ private void handlePartitionError(ShareAcknowledgeResponseData.PartitionData par
AtomicBoolean shouldRetry) {
if (partitionError.exception() != null) {
boolean retry = false;
- if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH) {
+ if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH || partitionError == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
Review Comment:
could we add unit tests to verify this?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -937,7 +938,7 @@ private void sendAcknowledgementsAndLeaveGroup(final Timer timer, final AtomicRe
// If users have fatal error, they will get some exceptions in the background queue.
// When running unsubscribe, these exceptions should be ignored, or users can't unsubscribe successfully.
processBackgroundEvents(unsubscribeEvent.future(), timer, e -> (e instanceof GroupAuthorizationException
- || e instanceof TopicAuthorizationException));
+ || e instanceof TopicAuthorizationException || e instanceof InvalidTopicException));
Review Comment:
same as above
--
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-19084: Port KAFKA-16224, KAFKA-16764 for ShareConsumers [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19369:
URL: https://github.com/apache/kafka/pull/19369#discussion_r2030220644
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -937,7 +938,7 @@ private void sendAcknowledgementsAndLeaveGroup(final Timer timer, final AtomicRe
// If users have fatal error, they will get some exceptions in the background queue.
// When running unsubscribe, these exceptions should be ignored, or users can't unsubscribe successfully.
processBackgroundEvents(unsubscribeEvent.future(), timer, e -> (e instanceof GroupAuthorizationException
- || e instanceof TopicAuthorizationException));
+ || e instanceof TopicAuthorizationException || e instanceof InvalidTopicException));
Review Comment:
Thanks, I have added 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-19084: Port KAFKA-16224, KAFKA-16764 for ShareConsumers [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19369:
URL: https://github.com/apache/kafka/pull/19369#discussion_r2030220618
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -971,9 +971,11 @@ private void handlePartitionError(ShareAcknowledgeResponseData.PartitionData par
AtomicBoolean shouldRetry) {
if (partitionError.exception() != null) {
boolean retry = false;
- if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH) {
+ if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH || partitionError == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
Review Comment:
Thanks, I have added 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-19084: Port KAFKA-16224, KAFKA-16764 for ShareConsumers [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #19369:
URL: https://github.com/apache/kafka/pull/19369
--
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