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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
