kirktrue commented on code in PR #20521:
URL: https://github.com/apache/kafka/pull/20521#discussion_r2411119422
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -369,17 +373,37 @@ void cleanup() {
* If there is a metadata error, complete all uncompleted events that
require subscription metadata.
*/
private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
- List<CompletableApplicationEvent<?>> subscriptionMetadataEvent = new
ArrayList<>();
+ List<MetadataErrorNotifiableEvent> notifiables = new ArrayList<>();
for (CompletableEvent<?> ce : events) {
- if (ce instanceof CompletableApplicationEvent &&
((CompletableApplicationEvent<?>) ce).requireSubscriptionMetadata())
- subscriptionMetadataEvent.add((CompletableApplicationEvent<?>)
ce);
+ if (ce instanceof MetadataErrorNotifiableEvent) {
+ notifiables.add((MetadataErrorNotifiableEvent) ce);
+ }
}
- if (subscriptionMetadataEvent.isEmpty())
+ if (notifiables.isEmpty())
return;
-
networkClientDelegate.getAndClearMetadataError().ifPresent(metadataError ->
- subscriptionMetadataEvent.forEach(event ->
event.future().completeExceptionally(metadataError))
- );
+
+ Optional<Exception> metadataErrorOpt =
networkClientDelegate.getAndClearMetadataError();
+
+ if (metadataErrorOpt.isEmpty())
+ return;
+
+ Exception metadataError = metadataErrorOpt.get();
Review Comment:
It's a little bit different because it needs to return a flag. I've tried to
simplify it as much as reasonably possible. PTAL at the latest. Thanks.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -369,17 +373,37 @@ void cleanup() {
* If there is a metadata error, complete all uncompleted events that
require subscription metadata.
*/
private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
- List<CompletableApplicationEvent<?>> subscriptionMetadataEvent = new
ArrayList<>();
+ List<MetadataErrorNotifiableEvent> notifiables = new ArrayList<>();
for (CompletableEvent<?> ce : events) {
- if (ce instanceof CompletableApplicationEvent &&
((CompletableApplicationEvent<?>) ce).requireSubscriptionMetadata())
- subscriptionMetadataEvent.add((CompletableApplicationEvent<?>)
ce);
+ if (ce instanceof MetadataErrorNotifiableEvent) {
+ notifiables.add((MetadataErrorNotifiableEvent) ce);
+ }
}
- if (subscriptionMetadataEvent.isEmpty())
+ if (notifiables.isEmpty())
return;
-
networkClientDelegate.getAndClearMetadataError().ifPresent(metadataError ->
- subscriptionMetadataEvent.forEach(event ->
event.future().completeExceptionally(metadataError))
- );
+
+ Optional<Exception> metadataErrorOpt =
networkClientDelegate.getAndClearMetadataError();
+
+ if (metadataErrorOpt.isEmpty())
+ return;
+
+ Exception metadataError = metadataErrorOpt.get();
+ notifiables.forEach(n ->
n.completeExceptionallyWithMetadataError(metadataError));
+ }
+
+ /**
+ * If there is a metadata error, complete this event exceptionally.
+ */
+ private boolean maybeFailOnMetadataError(MetadataErrorNotifiableEvent
notifiable) {
+ Optional<Exception> metadataErrorOpt =
networkClientDelegate.getAndClearMetadataError();
+
+ if (metadataErrorOpt.isEmpty())
+ return false;
+
+ Exception metadataError = metadataErrorOpt.get();
+ notifiable.completeExceptionallyWithMetadataError(metadataError);
+ return true;
Review Comment:
Done.
--
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]
