m1a2st commented on code in PR #17440:
URL: https://github.com/apache/kafka/pull/17440#discussion_r1860563297
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -325,4 +330,23 @@ void cleanup() {
log.debug("Closed the consumer network thread");
}
}
+
+ /**
+ * If there is a metadata error, complete the completable events which are
not passed by
+ * error event with the metadata error
+ * @param events the completable events
+ */
+ private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
+ if (networkClientDelegate.metadataError().isPresent()) {
+ Throwable metadataError =
networkClientDelegate.metadataError().get();
+ List<CompletableEvent<?>> completableApplicationEvent =
events.stream()
+ .filter(event -> !(event instanceof
CompletableApplicationEvent &&
+ ((CompletableApplicationEvent<?>)
event).isPassedByErrorEvent()))
+ .collect(Collectors.toList());
+ if (!completableApplicationEvent.isEmpty()) {
+ completableApplicationEvent.forEach(event ->
event.future().completeExceptionally(metadataError));
+ networkClientDelegate.clearMetadataError();
+ }
+ }
+ }
Review Comment:
If no events are present, we should not clear the metadata error. Instead,
we should propagate this error to the next event. Therefore, using `foreach` to
handle this scenario is not appropriate. we should check it have any event
first.
--
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
