chia7712 commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r3628557832
########## clients/src/main/java/org/apache/kafka/common/errors/BootstrapResolutionException.java: ########## @@ -0,0 +1,37 @@ +/* + * 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; +import org.apache.kafka.common.annotation.InterfaceAudience; + +/** + * 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 + */ [email protected] +public class BootstrapResolutionException extends KafkaException { + public BootstrapResolutionException(String message) { Review Comment: Have you considered adding it to consumer docs? ########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java: ########## @@ -2204,6 +2201,13 @@ private void acquireAndEnsureOpen() { release(); throw new IllegalStateException("This consumer has already been closed."); } + + try { + metadata.maybeThrowBootstrapFatalException(); Review Comment: Why is `metadata.maybeThrowBootstrapFatalException()`; only applied to the async consumer? Should we port it to the classic consumer? ########## clients/src/main/java/org/apache/kafka/clients/NetworkClient.java: ########## @@ -750,6 +803,18 @@ private void ensureActive() { public void close() { state.compareAndSet(State.ACTIVE, State.CLOSING); if (state.compareAndSet(State.CLOSING, State.CLOSED)) { + cancelBootstrapResolution(); + if (bootstrapExecutor != null) { + bootstrapExecutor.shutdownNow(); + // Wait briefly for the executor's worker thread to actually terminate. Otherwise + // a lingering thread can trip test-side thread-leak detectors even though the + // executor has already been shut down. + try { + bootstrapExecutor.awaitTermination(1, TimeUnit.SECONDS); Review Comment: Could you leverage `ThreadUtils.shutdownExecutorServiceQuietly`? ########## clients/src/main/java/org/apache/kafka/clients/NetworkClient.java: ########## @@ -1207,6 +1296,155 @@ 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()); + } + + // Start the timer on the first poll so its budget represents "time we spend on + // bootstrap since polling began" — the caller may have created the client well before + // its first API call, and we don't want that idle gap to eat into the budget. + if (bootstrapTimer == null) + bootstrapTimer = time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs); + + // 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 (maybeProcessBootstrapResolutionResult(currentTimeMs)) Review Comment: Pardon me, have you handled the "first poll" for timeout=0? It seems `checkBootstrapTimeout` still throws an exception directly when the first poll happens ########## test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/ClusterTestExtensions.java: ########## @@ -109,11 +109,16 @@ public class ClusterTestExtensions implements TestTemplateInvocationContextProvi private static final String PROCESS_REAPER_THREAD_PREFIX = "process reaper"; private static final String RMI_THREAD_PREFIX = "RMI"; private static final String JDK_INTERNAL_CLEANERIMPL_THREAD_PREFIX = "Cleaner-"; + // KIP-909 executor thread; daemon, per-client lifecycle. May briefly outlive the client's Review Comment: This is a bit weird to me, since we give a wait window for completion. If the thread is still alive after the timeout, it should be a true leak. -- 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]
