kirktrue commented on code in PR #17440:
URL: https://github.com/apache/kafka/pull/17440#discussion_r1805164231
##########
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:
For readability, could you introduce a `Predicate` variable, a la:
```suggestion
final Predicate<Exception> ignoreExceptions = e ->
e instanceof InvalidTopicException ||
e instanceof TopicAuthorizationException ||
e instanceof GroupAuthorizationException;
processBackgroundEvents(unsubscribeEvent.future(), timer,
ignoreExceptions);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1132,7 +1135,11 @@ private Map<TopicPartition, Long>
beginningOrEndOffset(Collection<TopicPartition
Map<TopicPartition, OffsetAndTimestampInternal>
offsetAndTimestampMap;
try {
- offsetAndTimestampMap =
applicationEventHandler.addAndGet(listOffsetsEvent);
+ applicationEventHandler.add(listOffsetsEvent);
+ offsetAndTimestampMap = processBackgroundEvents(
+ listOffsetsEvent.future(),
+ timer, __ -> false
+ );
Review Comment:
As I understand it, we need to check for the errors from the background
thread. But do we need to check _repeatedly_ during the execution of the
`ListOffsetsEvent`, or can we just check once beforehand?
```suggestion
processBackgroundEvents();
offsetAndTimestampMap =
applicationEventHandler.addAndGet(listOffsetsEvent);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1584,7 +1593,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(
+ checkAndUpdatePositionsEvent.future(),
+ timer, __ -> false
+ );
Review Comment:
```suggestion
processBackgroundEvents();
cachedSubscriptionHasAllFetchPositions =
applicationEventHandler.addAndGet(checkAndUpdatePositionsEvent);
```
--
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]
