chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3520934561
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1210,6 +1280,137 @@ 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()) {
Review Comment:
We could consolidate the condition into `processBootstrapResolutionResult`.
For example:
```java
/**
* Checks if there is a completed bootstrap DNS resolution and processes
the result.
* @return true if the client is now bootstrapped and the calling method
should early return.
*/
private boolean maybeProcessBootstrapResolutionResult(final long
currentTimeMs) {
if (pendingBootstrapResolution == null ||
!pendingBootstrapResolution.isDone())
return false;
List<InetSocketAddress> servers = List.of();
try {
servers = pendingBootstrapResolution.getNow(List.of());
} catch (CompletionException e) {
log.debug("DNS resolution failed", e);
}
pendingBootstrapResolution = null;
if (!servers.isEmpty()) {
log.debug("Bootstrap DNS resolution succeeded, {} servers
resolved", servers.size());
metadataUpdater.bootstrap(servers);
return true;
}
log.debug("Failed to resolve bootstrap servers, will retry after
{}ms. Remaining time: {}ms",
bootstrapConfiguration.retryBackoffMs,
bootstrapTimer.remainingMs());
bootstrapResolutionRetryMs = currentTimeMs +
bootstrapConfiguration.retryBackoffMs;
return false;
}
```
##########
clients/src/main/java/org/apache/kafka/common/errors/BootstrapResolutionException.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+import org.apache.kafka.common.KafkaException;
+
+/**
+ * Indicates that the {@link org.apache.kafka.clients.NetworkClient} was
unable to resolve a DNS address within
+ * the time specified by {@link
org.apache.kafka.clients.CommonClientConfigs#BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG}
+ *
+ * @see org.apache.kafka.clients.CommonClientConfigs
+ */
+public class BootstrapResolutionException extends KafkaException {
Review Comment:
Please add `@InterfaceAudience.Public`
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1210,6 +1280,137 @@ 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;
+ }
+
+ maybeStartBootstrapResolution(currentTimeMs);
Review Comment:
If `bootstrap.resolve.timeout.ms` is zero, we will cancel the resolution
task immediately. Should we give it a shot
##########
tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java:
##########
@@ -182,6 +183,11 @@ static AdminClient create(AbstractConfig config) {
"admin",
ClientUtils.createChannelBuilder(config, time, logContext),
logContext);
+ NetworkClient.BootstrapConfiguration bootstrapConfiguration =
NetworkClient.BootstrapConfiguration.enabled(
Review Comment:
Since this tool is being deprecated in #20598, I think we can leave it as is
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -360,6 +388,19 @@ public NetworkClient(MetadataUpdater metadataUpdater,
this.rebootstrapTriggerMs = rebootstrapTriggerMs;
this.metadataRecoveryStrategy = metadataRecoveryStrategy;
this.metadataClusterCheckEnable = metadataClusterCheckEnable;
+ this.bootstrapConfiguration = bootstrapConfiguration;
+ // Bootstrap timer is lazily initialized on first poll to ensure
timeout starts when polling begins
+ this.bootstrapTimer = null;
+ // Create executor for async DNS resolution if bootstrap is enabled
+ if (bootstrapConfiguration != BootstrapConfiguration.DISABLED) {
+ this.bootstrapExecutor = Executors.newSingleThreadExecutor(r -> {
Review Comment:
```java
this.bootstrapExecutor =
Executors.newSingleThreadExecutor(ThreadUtils.createThreadFactory("kafka-bootstrap-dns-resolver",
true));
```
--
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]
