This is an automated email from the ASF dual-hosted git repository.
frankvicky pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d1c0bd82c0e KAFKA-14648: Do not fail clients if bootstrap servers is
not immediately resolvable (2/2) (#22897)
d1c0bd82c0e is described below
commit d1c0bd82c0e6591ec999fb23e5feca05360e7256
Author: TengYao Chi <[email protected]>
AuthorDate: Wed Jul 29 21:00:39 2026 +0100
KAFKA-14648: Do not fail clients if bootstrap servers is not immediately
resolvable (2/2) (#22897)
This PR is a part of KIP-909. It aims to apply the KIP-909 to Share
Consumer.
Reviewers: Sushant Mahajan <[email protected]>, Lianet Magrans
<[email protected]>
---
.../consumer/internals/ShareConsumerImpl.java | 15 +++++--
.../clients/consumer/KafkaShareConsumerTest.java | 48 ++++++++++++++++++++++
2 files changed, 59 insertions(+), 4 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
index 04d3955ec02..a0c980a6cd2 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
@@ -55,6 +55,7 @@ import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.BootstrapResolutionException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.InvalidGroupIdException;
@@ -77,7 +78,6 @@ import org.apache.kafka.common.utils.internals.LogContext;
import org.slf4j.Logger;
import org.slf4j.event.Level;
-import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
@@ -272,8 +272,6 @@ public class ShareConsumerImpl<K, V> implements
ShareConsumerDelegate<K, V> {
metrics.reporters(),
Arrays.asList(deserializers.keyDeserializer(),
deserializers.valueDeserializer()));
this.metadata = new ShareConsumerMetadata(config, subscriptions,
logContext, clusterResourceListeners);
- final List<InetSocketAddress> addresses =
ClientUtils.parseAndValidateAddresses(config);
- metadata.bootstrap(addresses);
this.shareFetchMetricsManager =
createShareFetchMetricsManager(metrics);
ApiVersions apiVersions = new ApiVersions();
@@ -1095,8 +1093,11 @@ public class ShareConsumerImpl<K, V> implements
ShareConsumerDelegate<K, V> {
try {
// If users have fatal error, they will get some exceptions in the
background queue.
// When running unsubscribe, these exceptions should be ignored,
or users can't unsubscribe successfully.
+ // BootstrapResolutionException is also ignored here: a permanent
DNS failure prevents the
+ // unsubscribe from completing normally, but it is not actionable
during close.
processBackgroundEvents(unsubscribeEvent.future(), timer, e -> (e
instanceof GroupAuthorizationException
- || e instanceof TopicAuthorizationException || e instanceof
InvalidTopicException));
+ || e instanceof TopicAuthorizationException || e instanceof
InvalidTopicException
+ || e instanceof BootstrapResolutionException));
log.info("Completed releasing assignment and leaving group to
close consumer.");
} catch (TimeoutException e) {
log.warn("Consumer triggered an unsubscribe event to leave the
group but couldn't " +
@@ -1125,6 +1126,12 @@ public class ShareConsumerImpl<K, V> implements
ShareConsumerDelegate<K, V> {
release();
throw new IllegalStateException("This consumer has already been
closed.");
}
+ try {
+ metadata.maybeThrowBootstrapFatalException();
+ } catch (RuntimeException e) {
+ release();
+ throw e;
+ }
}
/**
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
index 45242f6d0b2..51d7d61f2b5 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.clients.consumer;
+import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.MockClient;
import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
@@ -27,6 +28,7 @@ import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.errors.BootstrapResolutionException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.message.ShareAcknowledgeResponseData;
import org.apache.kafka.common.message.ShareFetchResponseData;
@@ -66,7 +68,9 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
// This test exercises the KafkaShareConsumer with the MockClient to validate
the Kafka protocol RPCs
@Timeout(value = 120)
@@ -409,4 +413,48 @@ public class KafkaShareConsumerTest {
.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 {
+ consumer.close();
+ }
+ }
}
