lianetm commented on code in PR #22897:
URL: https://github.com/apache/kafka/pull/22897#discussion_r3676160340
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java:
##########
@@ -409,4 +414,56 @@ private ShareAcknowledgeResponse
shareAcknowledgeResponse(TopicIdPartition tip)
.setResponses(new
ShareAcknowledgeResponseData.ShareAcknowledgeTopicResponseCollection(List.of(topicResponse)))
);
}
+
+ @Test
+ public void
testShareConsumerBootstrapResolutionExceptionPropagatedToPoll() {
+ // Use an invalid hostname that will fail DNS resolution (using RFC
6761 reserved .invalid TLD)
+ String invalidHost = "unresolvable.invalid:9092";
+
+ Map<String, Object> configs = Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName(),
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName(),
+ CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, invalidHost,
+ // Set a short bootstrap timeout so the test doesn't take too long
+ CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG, "3000",
+ ConsumerConfig.GROUP_ID_CONFIG, "test-share-group"
+ );
+
+ KafkaShareConsumer<String, String> consumer = new
KafkaShareConsumer<>(configs);
+ try {
+ consumer.subscribe(Set.of("test-topic"));
+
+ // Poll continuously until we get the BootstrapResolutionException
+ // The exception should be thrown after
bootstrap.resolve.timeout.ms expires
+ BootstrapResolutionException exception =
+ assertThrows(BootstrapResolutionException.class, () -> {
+ long startTime = System.currentTimeMillis();
+ long maxWaitTime = 15000; // 15 seconds max to prevent
test hanging
+
+ while (System.currentTimeMillis() - startTime <
maxWaitTime) {
+ consumer.poll(Duration.ofMillis(100));
+ }
+ fail("Expected BootstrapResolutionException to be thrown
within " + maxWaitTime + "ms");
+ });
+
+ // Verify the exception message contains information about DNS
resolution failure
+ assertTrue(exception.getMessage().contains("Failed to resolve
bootstrap servers") ||
+ exception.getMessage().contains("DNS resolution"),
+ "Exception message should mention DNS resolution
failure: " + exception.getMessage());
+
+ // After the first failure, any further API call must also throw.
This guards against
+ // accidentally clearing the bootstrap error from the metadata
layer.
+ assertThrows(BootstrapResolutionException.class, () ->
consumer.poll(Duration.ofMillis(100)));
+ } finally {
+ // The graceful shutdown path itself triggers network I/O which
will re-observe the
+ // permanent bootstrap failure and wrap it in KafkaException.
That's expected for this
+ // test — we only care that the API-facing calls surface the
bootstrap error, not
+ // that close() also completes cleanly under a broken bootstrap.
Review Comment:
not sure about this, do we want to propagate this on close? Does the regular
consumer throw it?
I expect the async consumer doesn't (on close it only processes background
events, and metadata error do not come as one there). As for the classic, not
sure, but if let's say it wasn't able to resolve DNS and then close, even those
close does poll, I'm not sure it would propagate the error? (it only polls for
pending requests, which I imagine it wasn't able to generate if it didn't
resolve the brokers).
So to confirm the behaviour , but if the consumers do not throw this on
close as I imagine above, we shouldn't on the Share I would say, and it
actually already filters out exceptions here
https://github.com/apache/kafka/blob/74ef4dd97aa68ca30a3992f59d12fd0f7a0ff2ac/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L1098-L1099
--
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]
