lianetm commented on code in PR #17440:
URL: https://github.com/apache/kafka/pull/17440#discussion_r1875198457
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -325,4 +333,21 @@ void cleanup() {
log.debug("Closed the consumer network thread");
}
}
+
+ /**
+ * If there is a metadata error, complete all uncompleted events that
require subscription metadata.
+ */
+ private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
+ List<? extends CompletableApplicationEvent<?>>
subscriptionMetadataEvent = events.stream()
+ .filter(e -> e instanceof CompletableApplicationEvent<?>)
+ .map(e -> (CompletableApplicationEvent<?>) e)
+
.filter(CompletableApplicationEvent::requireSubscriptionMetadata)
+ .collect(Collectors.toUnmodifiableList());
Review Comment:
nit: `.toList();`? (I believe it's unmodifiable too)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -150,7 +155,11 @@ private void maybePropagateMetadataError() {
try {
metadata.maybeThrowAnyException();
Review Comment:
ok, we're saying it gets cleared as soon as it's propagated (to ensure that
it's indeed propagated), but we could still get an exception from a previously
sent request (this was my concern). But that's what we have with the classic
consumer actually, so it's consistent (there are actually tests specifically
covering the behaviour considering that)
https://github.com/apache/kafka/blob/ee4264439ddda7bdebcaa845752b824abba14161/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala#L553-L554
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegateTest.java:
##########
@@ -212,9 +213,26 @@ public void testPropagateMetadataError() {
AuthenticationException authException = new
AuthenticationException("Test Auth Exception");
doThrow(authException).when(metadata).maybeThrowAnyException();
+ NetworkClientDelegate networkClientDelegate =
newNetworkClientDelegate(false);
+ assertTrue(networkClientDelegate.getAndClearMetadataError().isEmpty());
+ networkClientDelegate.poll(0, time.milliseconds());
+
+ networkClientDelegate.getAndClearMetadataError().ifPresent(
Review Comment:
if the metadata error is not present this won't fail right, so should we
check it's present? (and then check the value)
--
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
