cadonna commented on code in PR #14879: URL: https://github.com/apache/kafka/pull/14879#discussion_r1413910032
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java: ########## @@ -131,6 +133,77 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> { private static final long NO_CURRENT_THREAD = -1L; + /** + * An {@link org.apache.kafka.clients.consumer.internals.events.EventProcessor} that is created and executes in the + * application thread for the purpose of processing {@link BackgroundEvent background events} generated by the + * {@link ConsumerNetworkThread network thread}. + * Those events are generally of two types: + * + * <ul> + * <li>Errors that occur in the network thread that need to be propagated to the application thread</li> + * <li>{@link ConsumerRebalanceListener} callbacks that are to be executed on the application thread</li> + * </ul> + */ + public class BackgroundEventProcessor extends EventProcessor<BackgroundEvent> { + + public BackgroundEventProcessor(final LogContext logContext, + final BlockingQueue<BackgroundEvent> backgroundEventQueue) { + super(logContext, backgroundEventQueue); + } + + /** + * Process the events—if any—that were produced by the {@link ConsumerNetworkThread network thread}. + * It is possible that {@link org.apache.kafka.clients.consumer.internals.events.ErrorBackgroundEvent an error} + * could occur when processing the events. In such cases, the processor will take a reference to the first + * error, continue to process the remaining events, and then throw the first error that occurred. + */ + @Override + public void process() { + AtomicReference<RuntimeException> firstError = new AtomicReference<>(); + process((event, error) -> firstError.compareAndSet(null, error)); + + if (firstError.get() != null) { + throw firstError.get(); + } + } + + @Override + public void process(final BackgroundEvent event) { + switch (event.type()) { + case ERROR: + process((ErrorBackgroundEvent) event); + break; + case GROUP_METADATA_UPDATE: + process((GroupMetadataUpdateEvent) event); + break; + default: + throw new IllegalArgumentException("Background event type " + event.type() + " was not expected"); + + } + } + + @Override + protected Class<BackgroundEvent> getEventClass() { + return BackgroundEvent.class; + } + + private void process(final ErrorBackgroundEvent event) { + throw event.error(); + } + + private void process(final GroupMetadataUpdateEvent event) { + if (AsyncKafkaConsumer.this.groupMetadata.isPresent()) { + final ConsumerGroupMetadata currentGroupMetadata = AsyncKafkaConsumer.this.groupMetadata.get(); + AsyncKafkaConsumer.this.groupMetadata = Optional.of(new ConsumerGroupMetadata( + event.groupId(), Review Comment: Yes, we do. We could verify that the group ID and the group instance ID are equal to catch bugs. Or we could just send the updateable fields to the application thread. ########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java: ########## @@ -131,6 +133,77 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> { private static final long NO_CURRENT_THREAD = -1L; + /** + * An {@link org.apache.kafka.clients.consumer.internals.events.EventProcessor} that is created and executes in the + * application thread for the purpose of processing {@link BackgroundEvent background events} generated by the + * {@link ConsumerNetworkThread network thread}. + * Those events are generally of two types: + * + * <ul> + * <li>Errors that occur in the network thread that need to be propagated to the application thread</li> + * <li>{@link ConsumerRebalanceListener} callbacks that are to be executed on the application thread</li> + * </ul> + */ + public class BackgroundEventProcessor extends EventProcessor<BackgroundEvent> { + + public BackgroundEventProcessor(final LogContext logContext, + final BlockingQueue<BackgroundEvent> backgroundEventQueue) { + super(logContext, backgroundEventQueue); + } + + /** + * Process the events—if any—that were produced by the {@link ConsumerNetworkThread network thread}. + * It is possible that {@link org.apache.kafka.clients.consumer.internals.events.ErrorBackgroundEvent an error} + * could occur when processing the events. In such cases, the processor will take a reference to the first + * error, continue to process the remaining events, and then throw the first error that occurred. + */ + @Override + public void process() { + AtomicReference<RuntimeException> firstError = new AtomicReference<>(); + process((event, error) -> firstError.compareAndSet(null, error)); + + if (firstError.get() != null) { + throw firstError.get(); + } + } + + @Override + public void process(final BackgroundEvent event) { + switch (event.type()) { + case ERROR: + process((ErrorBackgroundEvent) event); + break; + case GROUP_METADATA_UPDATE: + process((GroupMetadataUpdateEvent) event); + break; + default: + throw new IllegalArgumentException("Background event type " + event.type() + " was not expected"); + + } + } + + @Override + protected Class<BackgroundEvent> getEventClass() { + return BackgroundEvent.class; + } + + private void process(final ErrorBackgroundEvent event) { + throw event.error(); + } + + private void process(final GroupMetadataUpdateEvent event) { + if (AsyncKafkaConsumer.this.groupMetadata.isPresent()) { + final ConsumerGroupMetadata currentGroupMetadata = AsyncKafkaConsumer.this.groupMetadata.get(); + AsyncKafkaConsumer.this.groupMetadata = Optional.of(new ConsumerGroupMetadata( + event.groupId(), + event.memberEpoch(), + event.memberId() != null ? event.memberId() : currentGroupMetadata.memberId(), + event.groupInstanceId() Review Comment: See 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org