lianetm commented on code in PR #15742: URL: https://github.com/apache/kafka/pull/15742#discussion_r1568991555
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java: ########## @@ -228,13 +228,16 @@ private void process(final ErrorEvent event) { } private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) { - ApplicationEvent invokedEvent = invokeRebalanceCallbacks( + ConsumerRebalanceListenerCallbackCompletedEvent invokedEvent = invokeRebalanceCallbacks( rebalanceListenerInvoker, event.methodName(), event.partitions(), event.future() ); applicationEventHandler.add(invokedEvent); + if (invokedEvent.error().isPresent()) { + throw invokedEvent.error().get(); Review Comment: I was expecting to see here the logic for wrapping the callback error into a `KafkaException`, but I see it is at a lower level in the `invokeRebalanceCallbacks`, which it's a bit more obfuscated I would say? Still I see how it's deeply tied to the `ConsumerRebalanceListenerCallbackCompletedEvent` so ok for me to leave as it is if we feel it's clear enough. -- 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