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,44 @@ public final class ClientUtils {
     private ClientUtils() {
     }
 
+    public static List<InetSocketAddress> validateAddresses(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 (clientDnsLookup == ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+                InetAddress[] inetAddresses;
+                try {
+                    inetAddresses = InetAddress.getAllByName(host);
+                } catch (UnknownHostException e) {
+                    inetAddresses = new InetAddress[0];
+                }
+
+                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> parseAndValidateAddresses(AbstractConfig config) {
         List<String> urls = config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
         String clientDnsLookupConfig = config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG);
@@ -219,6 +257,7 @@ public static NetworkClient createNetworkClient(AbstractConfig config,
                                                     ClientTelemetrySender clientTelemetrySender) {
         ChannelBuilder channelBuilder = null;
         Selector selector = null;
+        NetworkClient.BootstrapConfiguration bootstrapConfiguration;
 
         try {
             channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext);
@@ -228,6 +267,20 @@ public static NetworkClient createNetworkClient(AbstractConfig config,
                     metricsGroupPrefix,
                     channelBuilder,
                     logContext);
+            List<String> bootstrapServers = config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+            ClientDnsLookup dnsLookup = ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG));
+
+            // Only validate if bootstrap servers are provided (non-empty list)
+            // This allows configurations that don't use bootstrap (e.g., broker-to-broker) to skip validation
+            if (bootstrapServers != null && !bootstrapServers.isEmpty()) {
+                parseAndValidateAddresses(bootstrapServers, dnsLookup);
+            }
+
+            bootstrapConfiguration = new NetworkClient.BootstrapConfiguration(
+                bootstrapServers,
+                dnsLookup,
+                CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS
+            );
             return new NetworkClient(metadataUpdater,
                     metadata,
                     selector,
@@ -248,7 +301,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/KafkaClient.java

@@ -212,5 +212,4 @@ ClientRequest newClientRequest(String nodeId,
      * was invoked for this client.
      */
     boolean active();
-
 }

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,16 @@ public void handleSuccessfulResponse(RequestHeader requestHeader, long now, Meta
         // Do nothing
     }
 
+    @Override
+    public boolean isBootstrapped() {
+        return false;
+    }
+
+    @Override
+    public void bootstrap(List<InetSocketAddress> addresses) {
+
+    }
+
     @Override
     public void close() {
     }

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

@@ -545,9 +545,13 @@ else if (metadata.error() == Errors.TOPIC_AUTHORIZATION_FAILED)
             return this.metadataSnapshot.mergeWith(metadataResponse.clusterId(), nodes, partitions,
                 unauthorizedTopics, invalidTopics, internalTopics, metadataResponse.controller(), topicIds,
                 (topic, isInternal) -> !topics.contains(topic) && retainTopic(topic, isInternal, nowMs));
-        else
+        else {
+            // Preserve the bootstrap flag from the current snapshot when creating a new one
+            boolean isBootstrapConfigured = this.metadataSnapshot.cluster().isBootstrapConfigured();
             return new MetadataSnapshot(metadataResponse.clusterId(), nodes, partitions,
-                unauthorizedTopics, invalidTopics, internalTopics, metadataResponse.controller(), topicIds);
+                unauthorizedTopics, invalidTopics, internalTopics, metadataResponse.controller(), topicIds,
+                isBootstrapConfigured, null);
+        }
     }
 
     /**

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

@@ -54,6 +54,7 @@ public class MetadataSnapshot {
     private final Map<TopicPartition, PartitionMetadata> metadataByPartition;
     private final Map<String, Uuid> topicIds;
     private final Map<Uuid, String> topicNames;
+    private final boolean isBootstrapConfigured;
     private Cluster clusterInstance;
 
     public MetadataSnapshot(String clusterId,
@@ -64,7 +65,7 @@ public MetadataSnapshot(String clusterId,
                   Set<String> internalTopics,
                   Node controller,
                   Map<String, Uuid> topicIds) {
-        this(clusterId, nodes, partitions, unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds, null);
+        this(clusterId, nodes, partitions, unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds, false, null);
     }
 
     // Visible for testing
@@ -76,6 +77,7 @@ public MetadataSnapshot(String clusterId,
         Set<String> internalTopics,
         Node controller,
         Map<String, Uuid> topicIds,
+        boolean isBootstrapConfigured,
         Cluster clusterInstance) {
         this.clusterId = clusterId;
         this.nodes = Collections.unmodifiableMap(nodes);
@@ -87,6 +89,7 @@ public MetadataSnapshot(String clusterId,
         this.topicNames = Collections.unmodifiableMap(
             topicIds.entrySet().stream().collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey))
         );
+        this.isBootstrapConfigured = isBootstrapConfigured;
 
         Map<TopicPartition, PartitionMetadata> tmpMetadataByPartition = new HashMap<>(partitions.size());
         for (PartitionMetadata p : partitions) {
@@ -199,8 +202,9 @@ MetadataSnapshot mergeWith(String newClusterId,
         Set<String> newInvalidTopics = fillSet(addInvalidTopics, invalidTopics, shouldRetainTopic);
         Set<String> newInternalTopics = fillSet(addInternalTopics, internalTopics, shouldRetainTopic);
 
+        // Preserve the bootstrap flag from the current snapshot during merge
         return new MetadataSnapshot(newClusterId, newNodes, newMetadataByPartition.values(), newUnauthorizedTopics,
-                newInvalidTopics, newInternalTopics, newController, newTopicIds);
+                newInvalidTopics, newInternalTopics, newController, newTopicIds, this.isBootstrapConfigured, null);
     }
 
     /**
@@ -227,8 +231,9 @@ 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);
+        // Use the factory method that preserves the bootstrap state
+        this.clusterInstance = Cluster.withBootstrapFlag(clusterId, nodes.values(), partitionInfos,
+                unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds, isBootstrapConfigured);
     }
 
     static MetadataSnapshot bootstrap(List<InetSocketAddress> addresses) {
@@ -240,12 +245,12 @@ static MetadataSnapshot bootstrap(List<InetSocketAddress> addresses) {
         }
         return new MetadataSnapshot(null, nodes, Collections.emptyList(),
                 Collections.emptySet(), Collections.emptySet(), Collections.emptySet(),
-                null, Collections.emptyMap(), Cluster.bootstrap(addresses));
+                null, Collections.emptyMap(), true, Cluster.bootstrap(addresses));
     }
 
     static MetadataSnapshot empty() {
         return new MetadataSnapshot(null, Collections.emptyMap(), Collections.emptyList(),
-                Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap(), Cluster.empty());
+                Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap(), false, Cluster.empty());
     }
 
     @Override

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.
      */