Connect to all DNS records of unresolved endpoint

netty bootstrap.connect uses only first address of unresolved InetSocketAddress.
That causes 4.x to not even try to connect to other when it first one fails.

This PR makes driver to resolve unresolved endpoint, instead of leaving
to to netty.
Making it possible to connect to any ip address from DNS contact endpoint.

Author: dkropachev

core/src/main/java/com/datastax/oss/driver/api/core/auth/PlainTextAuthProviderBase.java

@@ -21,14 +21,18 @@
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import com.datastax.oss.driver.api.core.session.Session;
 import com.datastax.oss.driver.shaded.guava.common.base.Charsets;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.nio.CharBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;
@@ -171,6 +175,12 @@ public SocketAddress resolve() {
             return new InetSocketAddress("127.0.0.1", 9042);
           }
 
+          @NonNull
+          @Override
+          public List<EndPoint> resolveAll() throws UnknownHostException {
+            return ImmutableList.of(this);
+          }
+
           @NonNull
           @Override
           public String asMetricPrefix() {

core/src/main/java/com/datastax/oss/driver/api/core/metadata/EndPoint.java

@@ -20,6 +20,8 @@
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.net.UnknownHostException;
+import java.util.List;
 
 /**
  * Encapsulates the information needed to open connections to a node.
@@ -40,6 +42,14 @@ public interface EndPoint {
   @NonNull
   SocketAddress resolve();
 
+  /**
+   * Resolves this instance to a socket address.
+   *
+   * <p>This will be called each time the driver opens a new connection to the node. The returned
+   * address cannot be null.
+   */
+  @NonNull
+  List<EndPoint> resolveAll() throws UnknownHostException;
   /**
    * Returns an alternate string representation for use in node-level metric names.
    *

core/src/main/java/com/datastax/oss/driver/internal/core/ContactPoints.java

@@ -19,20 +19,27 @@
 
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
+import com.datastax.oss.driver.internal.core.metadata.UnresolvedEndPoint;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet;
 import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
+
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/** Utility class to handle the initial contact points passed to the driver. */
+/**
+ * Utility class to handle the initial contact points passed to the driver.
+ */
 public class ContactPoints {
   private static final Logger LOG = LoggerFactory.getLogger(ContactPoints.class);
 
@@ -41,18 +48,17 @@ public static Set<EndPoint> merge(
 
     Set<EndPoint> result = Sets.newHashSet(programmaticContactPoints);
     for (String spec : configContactPoints) {
-      for (InetSocketAddress address : extract(spec, resolve)) {
-        DefaultEndPoint endPoint = new DefaultEndPoint(address);
+      for (EndPoint endPoint : extract(spec, resolve)) {
         boolean wasNew = result.add(endPoint);
         if (!wasNew) {
-          LOG.warn("Duplicate contact point {}", address);
+          LOG.warn("Duplicate contact point {}", spec);
         }
       }
     }
     return ImmutableSet.copyOf(result);
   }
 
-  private static Set<InetSocketAddress> extract(String spec, boolean resolve) {
+  private static Set<EndPoint> extract(String spec, boolean resolve) {
     int separator = spec.lastIndexOf(':');
     if (separator < 0) {
       LOG.warn("Ignoring invalid contact point {} (expecting host:port)", spec);
@@ -69,7 +75,7 @@ private static Set<InetSocketAddress> extract(String spec, boolean resolve) {
       return Collections.emptySet();
     }
     if (!resolve) {
-      return ImmutableSet.of(InetSocketAddress.createUnresolved(host, port));
+      return ImmutableSet.of(new UnresolvedEndPoint(host, port));
     } else {
       try {
         InetAddress[] inetAddresses = InetAddress.getAllByName(host);
@@ -79,9 +85,9 @@ private static Set<InetSocketAddress> extract(String spec, boolean resolve) {
               spec,
               Arrays.deepToString(inetAddresses));
         }
-        Set<InetSocketAddress> result = new HashSet<>();
+        Set<EndPoint> result = new HashSet<>();
         for (InetAddress inetAddress : inetAddresses) {
-          result.add(new InetSocketAddress(inetAddress, port));
+          result.add(new DefaultEndPoint(new InetSocketAddress(inetAddress, port)));
         }
         return result;
       } catch (UnknownHostException e) {

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPoint.java

@@ -18,9 +18,12 @@
 package com.datastax.oss.driver.internal.core.metadata;
 
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.Serializable;
 import java.net.InetSocketAddress;
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 
 public class DefaultEndPoint implements EndPoint, Serializable {
@@ -41,6 +44,12 @@ public InetSocketAddress resolve() {
     return address;
   }
 
+  @NonNull
+  @Override
+  public List<EndPoint> resolveAll() {
+    return ImmutableList.of(this);
+  }
+
   @Override
   public boolean equals(Object other) {
     if (other == this) {

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/InitialNodeListRefresh.java

@@ -25,10 +25,14 @@
 import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
+
+import java.net.UnknownHostException;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,8 +46,10 @@ class InitialNodeListRefresh extends NodesRefresh {
 
   private static final Logger LOG = LoggerFactory.getLogger(InitialNodeListRefresh.class);
 
-  @VisibleForTesting final Iterable<NodeInfo> nodeInfos;
-  @VisibleForTesting final Set<DefaultNode> contactPoints;
+  @VisibleForTesting
+  final Iterable<NodeInfo> nodeInfos;
+  @VisibleForTesting
+  final Set<DefaultNode> contactPoints;
 
   InitialNodeListRefresh(Iterable<NodeInfo> nodeInfos, Set<DefaultNode> contactPoints) {
     this.nodeInfos = nodeInfos;
@@ -72,21 +78,21 @@ public Result compute(
                 + "keeping only the first one",
             logPrefix,
             hostId);
+        continue;
+      }
+      EndPoint endPoint = nodeInfo.getEndPoint();
+      DefaultNode node = findIn(contactPoints, endPoint);
+      if (node == null) {
+        node = new DefaultNode(endPoint, context);
+        LOG.debug("[{}] Adding new node {}", logPrefix, node);
       } else {
-        EndPoint endPoint = nodeInfo.getEndPoint();
-        DefaultNode node = findIn(contactPoints, endPoint);
-        if (node == null) {
-          node = new DefaultNode(endPoint, context);
-          LOG.debug("[{}] Adding new node {}", logPrefix, node);
-        } else {
-          LOG.debug("[{}] Copying contact point {}", logPrefix, node);
-        }
-        if (tokenMapEnabled && tokenFactory == null && nodeInfo.getPartitioner() != null) {
-          tokenFactory = tokenFactoryRegistry.tokenFactoryFor(nodeInfo.getPartitioner());
-        }
-        copyInfos(nodeInfo, node, context);
-        newNodes.put(hostId, node);
+        LOG.debug("[{}] Copying contact point {}", logPrefix, node);
+      }
+      if (tokenMapEnabled && tokenFactory == null && nodeInfo.getPartitioner() != null) {
+        tokenFactory = tokenFactoryRegistry.tokenFactoryFor(nodeInfo.getPartitioner());
       }
+      copyInfos(nodeInfo, node, context);
+      newNodes.put(hostId, node);
     }
 
     ImmutableList.Builder<Object> eventsBuilder = ImmutableList.builder();

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java

@@ -44,17 +44,22 @@
 import com.datastax.oss.driver.internal.core.util.concurrent.Debouncer;
 import com.datastax.oss.driver.internal.core.util.concurrent.RunOrSchedule;
 import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import io.netty.util.concurrent.EventExecutor;
 import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+import jnr.ffi.annotations.Synchronized;
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -80,7 +85,8 @@ public class MetadataManager implements AsyncAutoCloseable {
   private volatile KeyspaceFilter keyspaceFilter;
   private volatile Boolean schemaEnabledProgrammatically;
   private volatile boolean tokenMapEnabled;
-  private volatile Set<DefaultNode> contactPoints;
+  private volatile Set<EndPoint> contactPoints;
+  private volatile Set<DefaultNode> resolvedContactPoints;
   private volatile boolean wasImplicitContactPoint;
   private volatile TypeCodec<TupleValue> tabletPayloadCodec = null;
 
@@ -102,7 +108,7 @@ protected MetadataManager(InternalDriverContext context, DefaultMetadata initial
             DefaultDriverOption.METADATA_SCHEMA_REFRESHED_KEYSPACES, Collections.emptyList());
     this.keyspaceFilter = KeyspaceFilter.newInstance(logPrefix, refreshedKeyspaces);
     this.tokenMapEnabled = config.getBoolean(DefaultDriverOption.METADATA_TOKEN_MAP_ENABLED);
-
+    this.resolvedContactPoints = new CopyOnWriteArraySet<>();
     context.getEventBus().register(ConfigChangeEvent.class, this::onConfigChanged);
   }
 
@@ -145,18 +151,19 @@ public void addContactPoints(Set<EndPoint> providedContactPoints) {
     // Convert the EndPoints to Nodes, but we can't put them into the Metadata yet, because we
     // don't know their host_id. So store them in a volatile field instead, they will get copied
     // during the first node refresh.
-    ImmutableSet.Builder<DefaultNode> contactPointsBuilder = ImmutableSet.builder();
+    ImmutableSet.Builder<EndPoint> contactPointsBuilder = ImmutableSet.builder();
     if (providedContactPoints == null || providedContactPoints.isEmpty()) {
       LOG.info(
           "[{}] No contact points provided, defaulting to {}", logPrefix, DEFAULT_CONTACT_POINT);
       this.wasImplicitContactPoint = true;
-      contactPointsBuilder.add(new DefaultNode(DEFAULT_CONTACT_POINT, context));
+      contactPointsBuilder.add(DEFAULT_CONTACT_POINT);
     } else {
       for (EndPoint endPoint : providedContactPoints) {
-        contactPointsBuilder.add(new DefaultNode(endPoint, context));
+        contactPointsBuilder.add(endPoint);
       }
     }
     this.contactPoints = contactPointsBuilder.build();
+    this.resolveContactPoints();
     LOG.debug("[{}] Adding initial contact points {}", logPrefix, contactPoints);
   }
 
@@ -167,7 +174,29 @@ public void addContactPoints(Set<EndPoint> providedContactPoints) {
    * @see #wasImplicitContactPoint()
    */
   public Set<DefaultNode> getContactPoints() {
-    return contactPoints;
+    return resolvedContactPoints;
+  }
+
+  @Synchronized
+  public void resolveContactPoints() {
+    ImmutableSet.Builder<EndPoint> resultBuilder = ImmutableSet.builder();
+    for (EndPoint endPoint : contactPoints) {
+      try {
+        resultBuilder.addAll(endPoint.resolveAll());
+      } catch (UnknownHostException e) {
+        LOG.error("failed to resolve contact endpoint {}", endPoint, e);
+      }
+    }
+
+    Set<EndPoint> result  = resultBuilder.build();
+    for (EndPoint endPoint : result) {
+      if (resolvedContactPoints.stream().anyMatch(resolved -> resolved.getEndPoint().equals(endPoint))) {
+        continue;
+      }
+      this.resolvedContactPoints.add(new DefaultNode(endPoint, context));
+    }
+
+    this.resolvedContactPoints.removeIf(endPoint -> !result.contains(endPoint.getEndPoint()));
   }
 
   /** Whether the default contact point was used (because none were provided explicitly). */
@@ -337,10 +366,13 @@ private SingleThreaded(InternalDriverContext context, DriverExecutionProfile con
     }
 
     private Void refreshNodes(Iterable<NodeInfo> nodeInfos) {
+      if (!didFirstNodeListRefresh) {
+        resolveContactPoints();
+      }
       MetadataRefresh refresh =
           didFirstNodeListRefresh
               ? new FullNodeListRefresh(nodeInfos)
-              : new InitialNodeListRefresh(nodeInfos, contactPoints);
+              : new InitialNodeListRefresh(nodeInfos, resolvedContactPoints);
       didFirstNodeListRefresh = true;
       return apply(refresh);
     }

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java

@@ -18,13 +18,15 @@
 package com.datastax.oss.driver.internal.core.metadata;
 
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.driver.shaded.guava.common.primitives.UnsignedBytes;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -72,6 +74,12 @@ public InetSocketAddress resolve() {
     }
   }
 
+  @NonNull
+  @Override
+  public List<EndPoint> resolveAll() {
+    return ImmutableList.of(this);
+  }
+
   @Override
   public boolean equals(Object other) {
     if (other == this) {