frankvicky commented on code in PR #22897:
URL: https://github.com/apache/kafka/pull/22897#discussion_r3676275815
##########
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:
Great catch — verified the classic & async consumers already handle this
cleanly
(`KafkaConsumerTest.testConsumerBootstrapResolutionExceptionPropagatedToPoll`
uses try-with-resources and `close()` doesn't throw), so Share was the odd one
out.
The Share close path hits `sendAcknowledgementsAndLeaveGroup` →
`processBackgroundEvents(unsubscribeEvent.future(), timer, filter)`, and the
filter only ignored `GroupAuthorizationException / TopicAuthorizationException
/ InvalidTopicException`, so a `BootstrapResolutionException` on the
unsubscribe future bubbled up and was captured into `firstException`, which
then produced the wrapping `KafkaException` at the end of close().
Added `BootstrapResolutionException` to that ignore predicate (matches the
intent you highlighted — the filter is exactly the right place). Now close()
completes cleanly under a broken bootstrap, and I removed the `try/catch`
wrapper from the test so it's consistent with the classic/async test's
try-with-resources pattern.
--
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]
