KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable

clients/src/main/java/org/apache/kafka/clients/ClientUtils.java

@@ -53,6 +53,72 @@ 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;
+        }
+        urls.forEach(url -> {
+            final String host = getHost(url);
+            final Integer port = getPort(url);
+
+            if (host == null || port == null) {
+                log.warn("Skipping invalid bootstrap URL: {}", url);
+                return;
+            }
+
+            try {
+                addresses.addAll(resolveAddress(url, host, port, clientDnsLookup));
+            } catch (UnknownHostException e) {
+                // Silently ignore - this matches the original behavior
+            }
+        });
+        return addresses;
+    }
+
     public static List<InetSocketAddress> parseAndValidateAddresses(AbstractConfig config) {
         List<String> urls = config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
         String clientDnsLookupConfig = config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG);
@@ -73,25 +139,7 @@ public static List<InetSocketAddress> parseAndValidateAddresses(List<String> url
                     if (host == null || port == null)
                         throw new ConfigException("Invalid url in " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
 
-                    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);
-                        }
-                    }
+                    addresses.addAll(resolveAddress(url, host, port, clientDnsLookup));
 
                 } catch (IllegalArgumentException e) {
                     throw new ConfigException("Invalid port in " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
@@ -151,6 +199,7 @@ static List<InetAddress> filterPreferredAddresses(InetAddress[] allAddresses) {
     }
 
     public static NetworkClient createNetworkClient(AbstractConfig config,
+                                                    List<String> bootstrapServers,
                                                     Metrics metrics,
                                                     String metricsGroupPrefix,
                                                     LogContext logContext,
@@ -161,6 +210,7 @@ public static NetworkClient createNetworkClient(AbstractConfig config,
                                                     Sensor throttleTimeSensor,
                                                     ClientTelemetrySender clientTelemetrySender) {
         return createNetworkClient(config,
+                bootstrapServers,
                 config.getString(CommonClientConfigs.CLIENT_ID_CONFIG),
                 metrics,
                 metricsGroupPrefix,
@@ -177,6 +227,7 @@ public static NetworkClient createNetworkClient(AbstractConfig config,
     }
 
     public static NetworkClient createNetworkClient(AbstractConfig config,
+                                                    List<String> bootstrapServers,
                                                     String clientId,
                                                     Metrics metrics,
                                                     String metricsGroupPrefix,
@@ -201,6 +252,14 @@ public static NetworkClient createNetworkClient(AbstractConfig config,
                     metricsGroupPrefix,
                     channelBuilder,
                     logContext);
+            ClientDnsLookup dnsLookup = ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG));
+
+            NetworkClient.BootstrapConfiguration bootstrapConfiguration = NetworkClient.BootstrapConfiguration.enabled(
+                bootstrapServers != null ? bootstrapServers : List.of(),
+                dnsLookup,
+                config.getLong(CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG),
+                config.getLong(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG));
+
             return new NetworkClient(metadataUpdater,
                     metadata,
                     selector,
@@ -221,7 +280,8 @@ public static NetworkClient createNetworkClient(AbstractConfig config,
                     hostResolver,
                     clientTelemetrySender,
                     config.getLong(CommonClientConfigs.METADATA_RECOVERY_REBOOTSTRAP_TRIGGER_MS_CONFIG),
-                    MetadataRecoveryStrategy.forName(config.getString(CommonClientConfigs.METADATA_RECOVERY_STRATEGY_CONFIG))
+                    MetadataRecoveryStrategy.forName(config.getString(CommonClientConfigs.METADATA_RECOVERY_STRATEGY_CONFIG)),
+                    bootstrapConfiguration
             );
         } catch (Throwable t) {
             closeQuietly(selector, "Selector");

clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java

@@ -60,6 +60,12 @@ public class CommonClientConfigs {
                                                        + "resolve each bootstrap address into a list of canonical names. After "
                                                        + "the bootstrap phase, this behaves the same as <code>use_all_dns_ips</code>.";
 
+    public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG = "bootstrap.resolve.timeout.ms";
+    public static final long DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS = 2 * 60 * 1000L;
+    public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_DOC = "Maximum amount of time clients can spend trying to" +
+        " resolve for the bootstrap server address. If the resolution cannot be completed within this timeframe, a " +
+        "<code>BootstrapResolutionException</code> will be thrown.";
+
     public static final String METADATA_MAX_AGE_CONFIG = "metadata.max.age.ms";
     public static final String METADATA_MAX_AGE_DOC = "The period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions.";
 

clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.common.requests.RequestHeader;
 
+import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
@@ -81,6 +82,19 @@ public void handleSuccessfulResponse(RequestHeader requestHeader, long now, Meta
         // Do nothing
     }
 
+    @Override
+    public boolean isBootstrapped() {
+        // ManualMetadataUpdater is designed for cases where nodes are manually set,
+        // so we consider it bootstrapped if nodes have been provided
+        return !nodes.isEmpty();
+    }
+
+    @Override
+    public void bootstrap(List<InetSocketAddress> addresses) {
+        // ManualMetadataUpdater doesn't use NetworkClient's bootstrap mechanism
+        // Nodes should be set manually via constructor or setNodes()
+    }
+
     @Override
     public void close() {
     }

clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java

@@ -227,8 +227,8 @@ private void computeClusterView() {
                 .stream()
                 .map(metadata -> MetadataResponse.toPartitionInfo(metadata, nodes))
                 .collect(Collectors.toList());
-        this.clusterInstance = new Cluster(clusterId, nodes.values(), partitionInfos, unauthorizedTopics,
-                invalidTopics, internalTopics, controller, topicIds);
+        this.clusterInstance = new Cluster(clusterId, nodes.values(), partitionInfos,
+                unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds);
     }
 
     static MetadataSnapshot bootstrap(List<InetSocketAddress> addresses) {

clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.requests.RequestHeader;
 
 import java.io.Closeable;
+import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Optional;
 
@@ -102,6 +103,17 @@ default boolean needsRebootstrap(long now, long rebootstrapTriggerMs) {
      */
     default void rebootstrap(long now) {}
 
+    /**
+     * Returns true if the metadata has been bootstrapped.
+     */
+    boolean isBootstrapped();
+
+    /**
+     * Bootstrap the metadata cache with the given addresses.
+     * @param addresses list of addresses for the bootstrap servers
+     */
+    void bootstrap(List<InetSocketAddress> addresses);
+
     /**
      * Close this updater.
      */