lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3462956747
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -4316,6 +4319,46 @@ public void
testConstructorFailsOnNetworkClientConstructorFailure(GroupProtocol
assertEquals("No LoginModule found for
org.example.InvalidLoginModule", cause.getMessage());
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void
testConsumerBootstrapResolutionExceptionPropagatedToPoll(GroupProtocol
protocol) throws InterruptedException {
+ // 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_PROTOCOL_CONFIG, protocol.name(),
+ ConsumerConfig.GROUP_ID_CONFIG, "test-group"
+ );
+
+ try (KafkaConsumer<String, String> consumer = new
KafkaConsumer<>(configs)) {
+ // Subscribe to a topic to trigger metadata fetch
+ 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, () -> {
Review Comment:
same here, let's assert we get the error in other API calls after this first
one that detects it
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -3416,4 +3424,26 @@ public static void resetCounters() {
CLOSE_COUNT.set(0);
}
}
+
+ @Test
+ public void testProducerBootstrapResolutionExceptionPropagated() {
Review Comment:
let's add another call to the producer API after the assertThrows, that will
ensure that all calls throw, not only the first one after the error (this is
what I expect should fail if we're not persisting the BootstrapException in the
metadata layer, comments above)
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -309,6 +309,14 @@ public void run() {
*
*/
void runOnce() {
+ try {
+ runOnceInternal();
+ } catch (BootstrapResolutionException e) {
+ metadata.fatalErrorAndNotify(e);
Review Comment:
with this we end up reusing the existing fatalError for the Bootstrap
exception, but that one gets cleared, which is not what we want (to ensure that
once Bootstrap fails, all API calls get the error, not just the first call).
What about the idea from before of having a separate bootstrapFatalError in
the metadata? that wouldn't get cleared, and we should set it at the network
client level where we get it probably (to ensure that it applies to all
consumer, producer, admin equally)
##########
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##########
@@ -53,6 +53,74 @@ public final class ClientUtils {
private ClientUtils() {
}
+ /**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+ private static List<InetSocketAddress> resolveAddress(
+ String url,
+ String host,
+ Integer port,
+ ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+ List<InetSocketAddress> addresses = new ArrayList<>();
+
+ if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+ InetAddress[] inetAddresses = InetAddress.getAllByName(host);
+ for (InetAddress inetAddress : inetAddresses) {
+ String resolvedCanonicalName =
inetAddress.getCanonicalHostName();
+ InetSocketAddress address = new
InetSocketAddress(resolvedCanonicalName, port);
+ if (address.isUnresolved()) {
+ log.warn("Couldn't resolve server {} from {} as DNS
resolution of the canonical hostname {} failed for {}",
+ url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
resolvedCanonicalName, host);
+ } else {
+ addresses.add(address);
+ }
+ }
+ } else {
+ InetSocketAddress address = new InetSocketAddress(host, port);
+ if (address.isUnresolved()) {
+ log.warn("Couldn't resolve server {} from {} as DNS resolution
failed for {}",
+ url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
host);
+ } else {
+ addresses.add(address);
+ }
+ }
+
+ return addresses;
+ }
+
+ public static List<InetSocketAddress> parseAddresses(List<String> urls,
ClientDnsLookup clientDnsLookup) {
+ List<InetSocketAddress> addresses = new ArrayList<>();
+ if (urls == null) {
+ return addresses;
+ }
+ for (String url : urls) {
+ if (Thread.currentThread().isInterrupted()) {
+ break;
+ }
+
+ final String host = getHost(url);
+ final Integer port = getPort(url);
+
+ if (host == null || port == null)
+ throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
Review Comment:
is this still needed in this async resolution? with the recent changes, it's
validated when creating the network client now, so we it be null at this point?
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1210,6 +1280,136 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+ public static class BootstrapConfiguration {
+ public static final BootstrapConfiguration DISABLED =
+ new BootstrapConfiguration(List.of(), null, 0, 0);
+
+ public final List<String> bootstrapServers;
+ public final ClientDnsLookup clientDnsLookup;
+ public final long bootstrapResolveTimeoutMs;
+ public final long retryBackoffMs;
+
+ private BootstrapConfiguration(final List<String> bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+ this.bootstrapServers = bootstrapServers;
+ this.clientDnsLookup = clientDnsLookup;
+ this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+ this.retryBackoffMs = retryBackoffMs;
+ }
+
+ public static BootstrapConfiguration enabled(final List<String>
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+ for (String url : bootstrapServers) {
+ if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+ throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+ }
+ return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+ }
+ }
+
+ /**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * <p>DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+ void ensureBootstrapped(final long currentTimeMs) {
+ if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+ return;
+
+ if (bootstrapException != null)
+ return;
+
+ if (Thread.interrupted()) {
+ cancelBootstrapResolution();
+ throw new InterruptException(new InterruptedException());
+ }
+
+ // Check if a pending resolution completed before checking the
timeout, so that a
+ // result arriving at the same time as the deadline is not incorrectly
rejected.
+ if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
+ processBootstrapResolutionResult(currentTimeMs);
+ if (metadataUpdater.isBootstrapped())
+ return;
+ }
+
+ if (bootstrapTimer != null) {
+ bootstrapTimer.update(currentTimeMs);
+ checkBootstrapTimeout();
+ }
+ maybeStartBootstrapResolution(currentTimeMs);
+ }
+
+ /**
+ * Check if bootstrap timeout has expired and throw exception if so.
+ */
+ private void checkBootstrapTimeout() {
+ if (bootstrapTimer.isExpired()) {
+ cancelBootstrapResolution();
+ bootstrapException = new BootstrapResolutionException("Failed to
resolve bootstrap servers after " +
+ bootstrapConfiguration.bootstrapResolveTimeoutMs + "ms. " +
+ "Please check your bootstrap.servers configuration and DNS
settings.");
+ throw bootstrapException;
Review Comment:
related to above, could we persist this "bootstrapFatalError" in metadata
here so that all API calls get it once it's discovered?
if we just throw here once, then early return on the ensureBootstraped, and
persist it only on the fatalError that get cleared, I guess it's just the first
API call after the failure that will get the exception.
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
Review Comment:
was this addressed? the java doc does not show the new exception
--
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]
