lianetm commented on code in PR #17440:
URL: https://github.com/apache/kafka/pull/17440#discussion_r1826189301
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1574,7 +1583,11 @@ private boolean updateFetchPositions(final Timer timer) {
try {
CheckAndUpdatePositionsEvent checkAndUpdatePositionsEvent = new
CheckAndUpdatePositionsEvent(calculateDeadlineMs(timer));
wakeupTrigger.setActiveTask(checkAndUpdatePositionsEvent.future());
- cachedSubscriptionHasAllFetchPositions =
applicationEventHandler.addAndGet(checkAndUpdatePositionsEvent);
+ applicationEventHandler.add(checkAndUpdatePositionsEvent);
+ cachedSubscriptionHasAllFetchPositions = processBackgroundEvents(
Review Comment:
I'm afraid that processing background here events could bring undesired
effects, mainly because it may include callbacks, which are totally unrelated
to a call to consumer.position() for instance.
If there's a call to consumer.position, and a reconciliation starts in the
background around the same time (ie. new partition assigned from the broker),
the background will enqueue a CallbackNeededEvent, so we could end up actually
running the callback here, as part of the call to position, which is not right.
I believe that we should not use `processBackgroundEvents` as a means of
knowing that a specific event that we sent to the background failed. We should
ensure that the event is completed exceptionally in the background instead.
Wouldn't that work, and avoid mixing errors and callbacks?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1277,10 +1284,12 @@ private void releaseAssignmentAndLeaveGroup(final Timer
timer) {
UnsubscribeEvent unsubscribeEvent = new
UnsubscribeEvent(calculateDeadlineMs(timer));
applicationEventHandler.add(unsubscribeEvent);
try {
- // If users subscribe to an invalid topic name, they will get
InvalidTopicException in error events,
+ // If users subscribe to an invalid topic name or subscribe an
authorization topic,
+ // they will get InvalidTopicException or
TopicAuthorizationException in error events,
// because network thread keeps trying to send MetadataRequest in
the background.
// Ignore it to avoid unsubscribe failed.
- processBackgroundEvents(unsubscribeEvent.future(), timer, e -> e
instanceof InvalidTopicException);
+ processBackgroundEvents(unsubscribeEvent.future(), timer,
+ e -> e instanceof InvalidTopicException || e instanceof
TopicAuthorizationException || e instanceof GroupAuthorizationException);
Review Comment:
uhm are we sure that swallowing TopicAuth and GroupAuth on close is the
right thing to do? I could surely be missing something, but I believe it's not
what the classic consumer does, see my comment on it on the other PR that is
also attempting this
https://github.com/apache/kafka/pull/17516#discussion_r1824955585
Thoughts?
--
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]
