[LI-CHERRY-PICK] Backport use all dns ips

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.clients;
 
 import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ClientDnsLookup;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.network.ChannelBuilder;
@@ -27,8 +28,11 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -89,4 +93,27 @@ public static ChannelBuilder createChannelBuilder(AbstractConfig config) {
         return ChannelBuilders.clientChannelBuilder(securityProtocol, JaasContext.Type.CLIENT, config, null,
                 clientSaslMechanism, true);
     }
+
+    static List<InetAddress> resolve(String host, ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+        InetAddress[] addresses = InetAddress.getAllByName(host);
+        if (ClientDnsLookup.USE_ALL_DNS_IPS == clientDnsLookup) {
+            return filterPreferredAddresses(addresses);
+        } else {
+            return Collections.singletonList(addresses[0]);
+        }
+    }
+
+    static List<InetAddress> filterPreferredAddresses(InetAddress[] allAddresses) {
+        List<InetAddress> preferredAddresses = new ArrayList<>();
+        Class<? extends InetAddress> clazz = null;
+        for (InetAddress address : allAddresses) {
+            if (clazz == null) {
+                clazz = address.getClass();
+            }
+            if (clazz.isInstance(address)) {
+                preferredAddresses.add(address);
+            }
+        }
+        return preferredAddresses;
+    }
 }

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

@@ -18,9 +18,16 @@
 
 import java.util.concurrent.ThreadLocalRandom;
 
+import org.apache.kafka.common.config.ClientDnsLookup;
 import org.apache.kafka.common.errors.AuthenticationException;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
 
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -33,8 +40,10 @@ final class ClusterConnectionStates {
     private final static int RECONNECT_BACKOFF_EXP_BASE = 2;
     private final double reconnectBackoffMaxExp;
     private final Map<String, NodeConnectionState> nodeState;
+    private final Logger log;
 
-    public ClusterConnectionStates(long reconnectBackoffMs, long reconnectBackoffMaxMs) {
+    public ClusterConnectionStates(long reconnectBackoffMs, long reconnectBackoffMaxMs, LogContext logContext) {
+        this.log = logContext.logger(ClusterConnectionStates.class);
         this.reconnectBackoffInitMs = reconnectBackoffMs;
         this.reconnectBackoffMaxMs = reconnectBackoffMaxMs;
         this.reconnectBackoffMaxExp = Math.log(this.reconnectBackoffMaxMs / (double) Math.max(reconnectBackoffMs, 1)) / Math.log(RECONNECT_BACKOFF_EXP_BASE);
@@ -101,25 +110,43 @@ public boolean isConnecting(String id) {
     }
 
     /**
-     * Enter the connecting state for the given connection.
+     * Enter the connecting state for the given connection, moving to a new resolved address if necessary.
      * @param id the id of the connection
-     * @param now the current time
+     * @param now the current time in ms
+     * @param host the host of the connection, to be resolved internally if needed
+     * @param clientDnsLookup the mode of DNS lookup to use when resolving the {@code host}
      */
-    public void connecting(String id, long now) {
-        if (nodeState.containsKey(id)) {
-            NodeConnectionState node = nodeState.get(id);
-            node.lastConnectAttemptMs = now;
-            node.state = ConnectionState.CONNECTING;
-        } else {
-            nodeState.put(id, new NodeConnectionState(ConnectionState.CONNECTING, now,
-                this.reconnectBackoffInitMs));
+    public void connecting(String id, long now, String host, ClientDnsLookup clientDnsLookup) {
+        NodeConnectionState connectionState = nodeState.get(id);
+        if (connectionState != null && connectionState.host().equals(host)) {
+            connectionState.lastConnectAttemptMs = now;
+            connectionState.state = ConnectionState.CONNECTING;
+            // Move to next resolved address, or if addresses are exhausted, mark node to be re-resolved
+            connectionState.moveToNextAddress();
+            return;
+        } else if (connectionState != null) {
+            log.info("Hostname for node {} changed from {} to {}.", id, connectionState.host(), host);
         }
+
+        // Create a new NodeConnectionState if nodeState does not already contain one
+        // for the specified id or if the hostname associated with the node id changed.
+        nodeState.put(id, new NodeConnectionState(ConnectionState.CONNECTING, now,
+            this.reconnectBackoffInitMs, host, clientDnsLookup));
+    }
+
+    /**
+     * Returns a resolved address for the given connection, resolving it if necessary.
+     * @param id the id of the connection
+     * @throws UnknownHostException if the address was not resolvable
+     */
+    public InetAddress currentAddress(String id) throws UnknownHostException {
+        return nodeState(id).currentAddress();
     }
 
     /**
      * Enter the disconnected state for the given node.
      * @param id the connection we have disconnected
-     * @param now the current time
+     * @param now the current time in ms
      */
     public void disconnected(String id, long now) {
         NodeConnectionState nodeState = nodeState(id);
@@ -194,7 +221,7 @@ public void ready(String id) {
     /**
      * Enter the authentication failed state for the given node.
      * @param id the connection identifier
-     * @param now the current time
+     * @param now the current time in ms
      * @param exception the authentication exception
      */
     public void authenticationFailed(String id, long now, AuthenticationException exception) {
@@ -209,7 +236,7 @@ public void authenticationFailed(String id, long now, AuthenticationException ex
      * Return true if the connection is in the READY state and currently not throttled.
      *
      * @param id the connection identifier
-     * @param now the current time
+     * @param now the current time in ms
      */
     public boolean isReady(String id, long now) {
         return isReady(nodeState.get(id), now);
@@ -223,7 +250,7 @@ private boolean isReady(NodeConnectionState state, long now) {
      * Return true if there is at least one node with connection in the READY state and not throttled. Returns false
      * otherwise.
      *
-     * @param now the current time
+     * @param now the current time in ms
      */
     public boolean hasReadyNodes(long now) {
         for (Map.Entry<String, NodeConnectionState> entry : nodeState.entrySet()) {
@@ -334,14 +361,55 @@ private static class NodeConnectionState {
         long reconnectBackoffMs;
         // Connection is being throttled if current time < throttleUntilTimeMs.
         long throttleUntilTimeMs;
+        private List<InetAddress> addresses;
+        private int addressIndex;
+        private final String host;
+        private final ClientDnsLookup clientDnsLookup;
 
-        public NodeConnectionState(ConnectionState state, long lastConnectAttempt, long reconnectBackoffMs) {
+        private NodeConnectionState(ConnectionState state, long lastConnectAttempt, long reconnectBackoffMs,
+                String host, ClientDnsLookup clientDnsLookup) {
             this.state = state;
+            this.addresses = Collections.emptyList();
+            this.addressIndex = -1;
             this.authenticationException = null;
             this.lastConnectAttemptMs = lastConnectAttempt;
             this.failedAttempts = 0;
             this.reconnectBackoffMs = reconnectBackoffMs;
             this.throttleUntilTimeMs = 0;
+            this.host = host;
+            this.clientDnsLookup = clientDnsLookup;
+        }
+
+        public String host() {
+            return host;
+        }
+
+        /**
+         * Fetches the current selected IP address for this node, resolving {@link #host()} if necessary.
+         * @return the selected address
+         * @throws UnknownHostException if resolving {@link #host()} fails
+         */
+        private InetAddress currentAddress() throws UnknownHostException {
+            if (addresses.isEmpty()) {
+                // (Re-)initialize list
+                addresses = ClientUtils.resolve(host, clientDnsLookup);
+                addressIndex = 0;
+            }
+
+            return addresses.get(addressIndex);
+        }
+
+        /**
+         * Jumps to the next available resolved address for this node. If no other addresses are available, marks the
+         * list to be refreshed on the next {@link #currentAddress()} call.
+         */
+        private void moveToNextAddress() {
+            if (addresses.isEmpty())
+                return; // Avoid div0. List will initialize on next currentAddress() call
+
+            addressIndex = (addressIndex + 1) % addresses.size();
+            if (addressIndex == 0)
+                addresses = Collections.emptyList(); // Exhausted list. Re-resolve on next currentAddress() call
         }
 
         public String toString() {

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

@@ -108,6 +108,10 @@ public class CommonClientConfigs {
         + "which will try to refresh metadata by choosing from existing resolved node set, this config will force resolving "
         + "the bootstrap url again to get new node set and use the new node set to send update metadata request";
 
+    public static final String CLIENT_DNS_LOOKUP_CONFIG = "client.dns.lookup";
+    public static final String CLIENT_DNS_LOOKUP_DOC = "<p>Controls how the client uses DNS lookups.</p><p>If set to <code>use_all_dns_ips</code> then, when the lookup returns multiple IP addresses for a hostname,"
+                                                        + " they will all be attempted to connect to before failing the connection. Applies to both bootstrap and advertised servers.</p>";
+
     /**
      * Postprocess the configuration so that exponential backoff is disabled when reconnect backoff
      * is explicitly configured but the maximum reconnect backoff is not explicitly configured.

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

@@ -155,8 +155,9 @@ public synchronized void incrementNodesTriedSinceLastSuccessfulRefresh() {
      * has been set by receiving stale metadata from a different cluster
      */
     public synchronized boolean shouldUpdateClusterMetadataFromBootstrap(long nowMs) {
-        return (this.nodesTriedSinceLastSuccessfulRefresh >= 1 &&
-            this.lastSuccessfulRefreshMs + this.maxClusterMetadataExpireTimeMs <= nowMs) ||
+        return this.maxClusterMetadataExpireTimeMs > 0 &&
+            (this.nodesTriedSinceLastSuccessfulRefresh >= 1 &&
+            (this.lastRefreshMs != 0 && this.maxClusterMetadataExpireTimeMs <= nowMs - this.lastSuccessfulRefreshMs)) ||
             this.forceClusterMetadataUpdateFromBootstrap;
     }
 

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

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ClientDnsLookup;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.memory.MemoryPool;
@@ -46,6 +47,7 @@
 import org.slf4j.Logger;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -100,6 +102,8 @@ public class NetworkClient implements KafkaClient {
     /* time in ms to wait before retrying to create connection to a server */
     private final long reconnectBackoffMs;
 
+    private final ClientDnsLookup clientDnsLookup;
+
     private final Time time;
 
     private boolean enableStickyMetadataFetch = true;
@@ -130,6 +134,7 @@ public NetworkClient(Selectable selector,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int defaultRequestTimeoutMs,
+                         ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
@@ -144,6 +149,7 @@ public NetworkClient(Selectable selector,
              socketSendBuffer,
              socketReceiveBuffer,
              defaultRequestTimeoutMs,
+             clientDnsLookup,
              time,
              discoverBrokerVersions,
              apiVersions,
@@ -161,6 +167,7 @@ public NetworkClient(Selectable selector,
             int socketSendBuffer,
             int socketReceiveBuffer,
             int defaultRequestTimeoutMs,
+            ClientDnsLookup clientDnsLookup,
             Time time,
             boolean discoverBrokerVersions,
             ApiVersions apiVersions,
@@ -177,6 +184,7 @@ public NetworkClient(Selectable selector,
              socketSendBuffer,
              socketReceiveBuffer,
              defaultRequestTimeoutMs,
+             clientDnsLookup,
              time,
              discoverBrokerVersions,
              apiVersions,
@@ -194,6 +202,7 @@ public NetworkClient(Selectable selector,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int defaultRequestTimeoutMs,
+                         ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
@@ -208,6 +217,7 @@ public NetworkClient(Selectable selector,
              socketSendBuffer,
              socketReceiveBuffer,
              defaultRequestTimeoutMs,
+             clientDnsLookup,
              time,
              discoverBrokerVersions,
              apiVersions,
@@ -225,6 +235,7 @@ public NetworkClient(Selectable selector,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int defaultRequestTimeoutMs,
+                         ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
@@ -240,6 +251,7 @@ public NetworkClient(Selectable selector,
              socketSendBuffer,
              socketReceiveBuffer,
              defaultRequestTimeoutMs,
+             clientDnsLookup,
              time,
              discoverBrokerVersions,
              apiVersions,
@@ -258,6 +270,7 @@ private NetworkClient(MetadataUpdater metadataUpdater,
                           int socketSendBuffer,
                           int socketReceiveBuffer,
                           int defaultRequestTimeoutMs,
+                          ClientDnsLookup clientDnsLookup,
                           Time time,
                           boolean discoverBrokerVersions,
                           ApiVersions apiVersions,
@@ -278,7 +291,7 @@ private NetworkClient(MetadataUpdater metadataUpdater,
         this.selector = selector;
         this.clientId = clientId;
         this.inFlightRequests = new InFlightRequests(maxInFlightRequestsPerConnection);
-        this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax);
+        this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax, logContext);
         this.socketSendBuffer = socketSendBuffer;
         this.socketReceiveBuffer = socketReceiveBuffer;
         this.correlation = 0;
@@ -291,6 +304,7 @@ private NetworkClient(MetadataUpdater metadataUpdater,
         this.throttleTimeSensor = throttleTimeSensor;
         this.logContext = logContext;
         this.log = logContext.logger(NetworkClient.class);
+        this.clientDnsLookup = clientDnsLookup;
         this.bootstrapServers.addAll(bootstrapServersConfig);
     }
 
@@ -952,22 +966,25 @@ private static void correlate(RequestHeader requestHeader, ResponseHeader respon
 
     /**
      * Initiate a connection to the given node
+     * @param node the node to connect to
+     * @param now current time in epoch milliseconds
      */
     private void initiateConnect(Node node, long now) {
         String nodeConnectionId = node.idString();
         try {
-            log.debug("Initiating connection to node {}", node);
-            this.connectionStates.connecting(nodeConnectionId, now);
+            connectionStates.connecting(nodeConnectionId, now, node.host(), clientDnsLookup);
+            InetAddress address = connectionStates.currentAddress(nodeConnectionId);
+            log.debug("Initiating connection to node {} using address {}", node, address);
             selector.connect(nodeConnectionId,
-                             new InetSocketAddress(node.host(), node.port()),
-                             this.socketSendBuffer,
-                             this.socketReceiveBuffer);
+                    new InetSocketAddress(address, node.port()),
+                    this.socketSendBuffer,
+                    this.socketReceiveBuffer);
         } catch (IOException e) {
+            log.warn("Error connecting to node {}", node, e);
             /* attempt failed, we'll try again after the backoff */
             connectionStates.disconnected(nodeConnectionId, now);
             /* maybe the problem is our metadata, update it */
             metadataUpdater.requestUpdate();
-            log.warn("Error connecting to node {}", node, e);
         }
     }