fix: expand contact point hostnames to all DNS IPs at connection time (DRIVER-201)

Addresses the initial-contact-endpoints aspect of DRIVER-201.

Problem: with RESOLVE_CONTACT_POINTS=false (the default), a contact
point hostname was stored as a single unresolved InetSocketAddress.
At connection time the load-balancing query plan contained exactly one
Node per hostname, so only the first IP returned by DNS was ever tried.
If that IP was non-responsive the driver raised AllNodesFailedException
with no fallback to other IPs the hostname might resolve to.

Solution (per @dkropachev's architectural direction):
- Deprecate DefaultDriverOption.RESOLVE_CONTACT_POINTS. Contact points
  are now always kept as unresolved hostnames (resolve=false is hardcoded
  in SessionBuilder), deferring DNS expansion to connection time.
- Add MetadataManager.getResolvedContactPoints(): for each contact point
  backed by an unresolved hostname it calls InetAddress.getAllByName() to
  expand the hostname to all known IPs, creating a synthetic DefaultNode
  for each IP. Already-resolved or non-InetSocketAddress endpoints pass
  through unchanged.
- LoadBalancingPolicyWrapper now calls getResolvedContactPoints() instead
  of getContactPoints() in newQueryPlan() (BEFORE/DURING_INIT states) and
  newControlReconnectionQueryPlan(), so the query plan contains one node
  per resolved IP and the driver naturally falls back to the next IP when
  one is unreachable.

Tests:
- 4 new MetadataManagerTest cases covering null state, already-resolved
  passthrough, single-hostname expansion, and multi-endpoint expansion.
- LoadBalancingPolicyWrapperTest updated to stub getResolvedContactPoints().
- New MockResolverIT.should_connect_when_first_dns_entry_is_non_responsive
  integration test: first DNS entry is a non-existent IP, session must
  open successfully against the remaining real IPs.

Author: nikagra

core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java

@@ -837,7 +837,11 @@ public enum DefaultDriverOption implements DriverOption {
    * Whether to resolve the addresses passed to `basic.contact-points`.
    *
    * <p>Value-type: boolean
+   *
+   * @deprecated Contact points are now always kept as unresolved hostnames and expanded to all
+   *     their DNS-mapped IPs lazily at connection time. Setting this option has no effect.
    */
+  @Deprecated
   RESOLVE_CONTACT_POINTS("advanced.resolve-contact-points"),
 
   /**

core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java

@@ -656,7 +656,13 @@ public String toString() {
   /** The coalescer reschedule interval. */
   public static final TypedDriverOption<Duration> COALESCER_INTERVAL =
       new TypedDriverOption<>(DefaultDriverOption.COALESCER_INTERVAL, GenericType.DURATION);
-  /** Whether to resolve the addresses passed to `basic.contact-points`. */
+  /**
+   * Whether to resolve the addresses passed to `basic.contact-points`.
+   *
+   * @deprecated Contact points are now always kept as unresolved hostnames and expanded to all
+   *     their DNS-mapped IPs lazily at connection time. Setting this option has no effect.
+   */
+  @Deprecated
   public static final TypedDriverOption<Boolean> RESOLVE_CONTACT_POINTS =
       new TypedDriverOption<>(DefaultDriverOption.RESOLVE_CONTACT_POINTS, GenericType.BOOLEAN);
   /**

core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java

@@ -935,11 +935,10 @@ protected final CompletionStage<CqlSession> buildDefaultSessionAsync() {
         programmaticArguments = programmaticArgumentsBuilder.build();
       }
 
-      boolean resolveAddresses =
-          defaultConfig.getBoolean(DefaultDriverOption.RESOLVE_CONTACT_POINTS, false);
-
+      // RESOLVE_CONTACT_POINTS is deprecated: contact points are always kept as unresolved
+      // hostnames and expanded to all their DNS IPs lazily at connection time.
       Set<EndPoint> contactPoints =
-          ContactPoints.merge(programmaticContactPoints, configContactPoints, resolveAddresses);
+          ContactPoints.merge(programmaticContactPoints, configContactPoints, false);
 
       if (keyspace == null && defaultConfig.isDefined(DefaultDriverOption.SESSION_KEYSPACE)) {
         keyspace =

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

@@ -147,8 +147,9 @@ public Queue<Node> newQueryPlan(
     switch (stateRef.get()) {
       case BEFORE_INIT:
       case DURING_INIT:
-        // The contact points are not stored in the metadata yet:
-        List<Node> nodes = new ArrayList<>(context.getMetadataManager().getContactPoints());
+        // The contact points are not stored in the metadata yet; use the expanded list so that
+        // all DNS-mapped IPs are tried (not just the first one per hostname).
+        List<Node> nodes = new ArrayList<>(context.getMetadataManager().getResolvedContactPoints());
         Collections.shuffle(nodes);
         return new ConcurrentLinkedQueue<>(nodes);
       case RUNNING:
@@ -170,11 +171,10 @@ public Queue<Node> newControlReconnectionQueryPlan() {
         .getConfig()
         .getDefaultProfile()
         .getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS)) {
-      Set<DefaultNode> originalNodes = context.getMetadataManager().getContactPoints();
-      List<Node> contactNodes = new ArrayList<>();
-      for (DefaultNode node : originalNodes) {
-        contactNodes.add(DefaultNode.newContactPoint(node.getEndPoint(), context));
-      }
+      // Use the DNS-expanded contact points so that all IPs for each hostname are appended as
+      // fallback candidates, not just the one that was stored when the session was built.
+      List<Node> contactNodes =
+          new ArrayList<>(context.getMetadataManager().getResolvedContactPoints());
       Collections.shuffle(contactNodes);
       // Append contact points to the end of the regular query plan so they serve as a fallback
       regularQueryPlan.addAll(contactNodes);

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

@@ -49,8 +49,11 @@
 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.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -173,6 +176,59 @@ public Set<DefaultNode> getContactPoints() {
     return contactPoints;
   }
 
+  /**
+   * Returns the contact points expanded to all their DNS-resolved IPs.
+   *
+   * <p>For each contact point whose underlying address is an unresolved hostname (i.e. stored as
+   * {@code InetSocketAddress.createUnresolved(...)} when {@code RESOLVE_CONTACT_POINTS=false}),
+   * this method calls {@link InetAddress#getAllByName(String)} to obtain every IP the hostname maps
+   * to and creates a synthetic contact-point {@link DefaultNode} for each IP. This lets the load
+   * balancing policy iterate over all candidate IPs rather than only the first one, so that a
+   * non-responsive IP does not block initial connection or control-connection reconnection.
+   *
+   * <p>Already-resolved addresses and non-{@link InetSocketAddress} endpoints are returned as-is.
+   */
+  public List<Node> getResolvedContactPoints() {
+    Set<DefaultNode> nodes = contactPoints;
+    if (nodes == null) {
+      return new ArrayList<>();
+    }
+    List<Node> result = new ArrayList<>();
+    for (DefaultNode node : nodes) {
+      EndPoint endPoint = node.getEndPoint();
+      if (endPoint instanceof DefaultEndPoint) {
+        InetSocketAddress address = ((DefaultEndPoint) endPoint).resolve();
+        if (address.isUnresolved()) {
+          // Expand hostname to all IPs so callers can try each one in turn.
+          try {
+            InetAddress[] all = InetAddress.getAllByName(address.getHostString());
+            if (all.length > 1) {
+              LOG.debug(
+                  "[{}] Contact point {} expands to {} addresses",
+                  logPrefix,
+                  address.getHostString(),
+                  all.length);
+            }
+            for (InetAddress ip : all) {
+              InetSocketAddress resolved = new InetSocketAddress(ip, address.getPort());
+              result.add(DefaultNode.newContactPoint(new DefaultEndPoint(resolved), context));
+            }
+          } catch (UnknownHostException e) {
+            LOG.warn(
+                "[{}] Could not resolve contact point hostname {}, skipping",
+                logPrefix,
+                address.getHostString(),
+                e);
+          }
+          continue;
+        }
+      }
+      // Already resolved or non-InetSocketAddress endpoint — use as-is.
+      result.add(node);
+    }
+    return result;
+  }
+
   /** Whether the default contact point was used (because none were provided explicitly). */
   public boolean wasImplicitContactPoint() {
     return wasImplicitContactPoint;

core/src/test/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapperTest.java

@@ -100,7 +100,8 @@ public void setup() {
             Objects.requireNonNull(node3.getHostId()), node3);
     when(metadataManager.getMetadata()).thenReturn(metadata);
     when(metadata.getNodes()).thenReturn(allNodes);
-    when(metadataManager.getContactPoints()).thenReturn(contactPoints);
+    when(metadataManager.getResolvedContactPoints())
+        .thenReturn(ImmutableList.copyOf(contactPoints));
     when(context.getMetadataManager()).thenReturn(metadataManager);
 
     when(context.getConfig()).thenReturn(config);
@@ -204,8 +205,7 @@ public void should_fetch_control_connection_query_plan_from_policy_after_init()
     assertThat(queryPlan.poll()).isEqualTo(node3);
     assertThat(queryPlan.poll()).isEqualTo(node2);
     assertThat(queryPlan.poll()).isEqualTo(node1);
-    // Remaining nodes are contact points appended at the end.
-    // They are new DefaultNode instances created via newContactPoint, so compare by endpoint.
+    // Remaining nodes are the resolved contact points appended at the end.
     Set<EndPoint> remainingEndpoints = new java.util.HashSet<>();
     for (Node n : queryPlan) {
       remainingEndpoints.add(n.getEndPoint());

core/src/test/java/com/datastax/oss/driver/internal/core/metadata/MetadataManagerTest.java

@@ -490,6 +490,66 @@ public void should_throw_on_registerNode_with_null_hostId() {
         .hasMessageContaining("Cannot register node without hostId");
   }
 
+  @Test
+  public void should_return_empty_list_when_contact_points_not_yet_set() {
+    // contactPoints field is null until addContactPoints is called
+    assertThat(metadataManager.getResolvedContactPoints()).isEmpty();
+  }
+
+  @Test
+  public void should_return_already_resolved_contact_points_unchanged() {
+    // Given — a contact point with an already-resolved InetSocketAddress
+    metadataManager.addContactPoints(ImmutableSet.of(END_POINT2));
+
+    // When
+    List<Node> resolved = metadataManager.getResolvedContactPoints();
+
+    // Then — the single node is returned as-is (no expansion needed)
+    assertThat(resolved).hasSize(1);
+    assertThat(resolved.get(0).getEndPoint()).isEqualTo(END_POINT2);
+  }
+
+  @Test
+  public void should_expand_unresolved_hostname_to_all_ips() {
+    // Given — a contact point with an unresolved hostname (localhost → 127.0.0.1)
+    EndPoint unresolvedEndPoint =
+        new DefaultEndPoint(InetSocketAddress.createUnresolved("localhost", 9042));
+    metadataManager.addContactPoints(ImmutableSet.of(unresolvedEndPoint));
+
+    // When
+    List<Node> resolved = metadataManager.getResolvedContactPoints();
+
+    // Then — at least one node is returned, each with a resolved address
+    assertThat(resolved).isNotEmpty();
+    for (Node node : resolved) {
+      InetSocketAddress addr = (InetSocketAddress) node.getEndPoint().resolve();
+      assertThat(addr.isUnresolved()).isFalse();
+      assertThat(addr.getPort()).isEqualTo(9042);
+    }
+  }
+
+  @Test
+  public void should_expand_multiple_contact_points_independently() {
+    // Given — two contact points: one already resolved, one unresolved
+    EndPoint resolvedEndPoint = END_POINT3;
+    EndPoint unresolvedEndPoint =
+        new DefaultEndPoint(InetSocketAddress.createUnresolved("localhost", 9042));
+    metadataManager.addContactPoints(ImmutableSet.of(resolvedEndPoint, unresolvedEndPoint));
+
+    // When
+    List<Node> resolved = metadataManager.getResolvedContactPoints();
+
+    // Then — at least 2 nodes: 1 for the resolved + at least 1 for localhost expansion
+    assertThat(resolved.size()).isGreaterThanOrEqualTo(2);
+    // The resolved endpoint must appear
+    assertThat(resolved).anySatisfy(n -> assertThat(n.getEndPoint()).isEqualTo(resolvedEndPoint));
+    // All returned addresses must be resolved
+    for (Node node : resolved) {
+      InetSocketAddress addr = (InetSocketAddress) node.getEndPoint().resolve();
+      assertThat(addr.isUnresolved()).isFalse();
+    }
+  }
+
   private static class TestMetadataManager extends MetadataManager {
 
     private List<MetadataRefresh> refreshes = new CopyOnWriteArrayList<>();

integration-tests/src/test/java/com/datastax/oss/driver/core/resolver/MockResolverIT.java

@@ -201,6 +201,46 @@ public void run_replace_test_20_times() {
     }
   }
 
+  /**
+   * Verifies that the driver can connect to a cluster when the first DNS entry for the contact
+   * point hostname resolves to a non-responsive IP address (DRIVER-201).
+   *
+   * <p>Contact points are always kept as unresolved hostnames and expanded to all their DNS IPs at
+   * connection time. When the first IP is unreachable the driver tries the remaining ones, so the
+   * session opens successfully even if the first DNS entry is non-responsive.
+   */
+  @Test
+  public void should_connect_when_first_dns_entry_is_non_responsive() {
+    // Use a 2-node cluster on 127.0.1.x.  Node 11 (127.0.1.11) does not exist and is therefore
+    // non-responsive; nodes 1 and 2 are real.
+    try (CcmBridge ccmBridge = CcmBridge.builder().withNodes(2).withIpPrefix("127.0.1.").build()) {
+      MultimapHostResolverProvider.removeResolverEntries("test.cluster.fake");
+      // First entry intentionally points to a non-existent/non-responsive address.
+      MultimapHostResolverProvider.addResolverEntry("test.cluster.fake", "127.0.1.11");
+      MultimapHostResolverProvider.addResolverEntry(
+          "test.cluster.fake", ccmBridge.getNodeIpAddress(1));
+      MultimapHostResolverProvider.addResolverEntry(
+          "test.cluster.fake", ccmBridge.getNodeIpAddress(2));
+      ccmBridge.create();
+      ccmBridge.start();
+
+      DriverConfigLoader loader =
+          new DefaultProgrammaticDriverConfigLoaderBuilder()
+              .withBoolean(TypedDriverOption.RECONNECT_ON_INIT.getRawOption(), false)
+              .withStringList(
+                  TypedDriverOption.CONTACT_POINTS.getRawOption(),
+                  Collections.singletonList("test.cluster.fake:9042"))
+              .build();
+
+      // The session must open successfully despite the first DNS entry being unreachable.
+      try (CqlSession session = new CqlSessionBuilder().withConfigLoader(loader).build()) {
+        ResultSet rs = session.execute("select * from system.local where key='local'");
+        assertThat(rs.one()).isNotNull();
+        waitForAllNodesUp(session, 2);
+      }
+    }
+  }
+
   // This is too long to run during CI, but is useful for manual investigations.
   @SuppressWarnings("unused")
   public void cannot_reconnect_with_resolved_socket() {