fix: connect via all DNS IPs when first contact point is non-responsive (DRIVER-201)

When RESOLVE_CONTACT_POINTS=false (the default), a hostname is stored as a
single unresolved InetSocketAddress. At connection time Netty's bootstrap
called InetAddress.getByName(), returning only the first IP. If that IP was
non-responsive the driver raised AllNodesFailedException with no fallback.

Refactor ChannelFactory to:
- Introduce ConnectRequest parameter object, eliminating the 9-parameter
  private connect() method.
- Decompose the connection logic into three single-responsibility methods:
  connect(ConnectRequest)      – resolves hostname via getAllByName() and
                                 builds a candidate list of all IPs;
  tryNextAddress(...)          – iterates candidates, falling back to the
                                 next IP on failure;
  connectToAddress(...)        – performs a single Netty connect and handles
                                 protocol-version negotiation on the same IP.
- DNS resolution is still deferred to connection time, preserving the
  dynamic-DNS semantics of RESOLVE_CONTACT_POINTS=false.

Add integration test should_connect_when_first_dns_entry_is_non_responsive
to MockResolverIT that maps the first DNS entry to a non-existent IP and
asserts the session opens successfully via the subsequent entries.

Author: nikagra

core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelFactory.java

@@ -55,9 +55,13 @@
 import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelPipeline;
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.ServerSocket;
 import java.net.SocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -196,69 +200,173 @@ CompletionStage<DriverChannel> connect(
     }
 
     connect(
-        endPoint,
-        shardingInfo,
-        shardId,
-        options,
-        nodeMetricUpdater,
-        currentVersion,
-        isNegotiating,
-        attemptedVersions,
-        resultFuture);
+        new ConnectRequest(
+            endPoint,
+            shardingInfo,
+            shardId,
+            options,
+            nodeMetricUpdater,
+            currentVersion,
+            isNegotiating,
+            attemptedVersions,
+            resultFuture));
     return resultFuture;
   }
 
-  private void connect(
-      EndPoint endPoint,
-      NodeShardingInfo shardingInfo,
-      Integer shardId,
-      DriverChannelOptions options,
-      NodeMetricUpdater nodeMetricUpdater,
-      ProtocolVersion currentVersion,
-      boolean isNegotiating,
-      List<ProtocolVersion> attemptedVersions,
-      CompletableFuture<DriverChannel> resultFuture) {
+  /**
+   * Bundles all per-connection-attempt state so it can be threaded through the decomposed connect
+   * methods without a growing parameter list.
+   */
+  private static class ConnectRequest {
+    final EndPoint endPoint;
+    final NodeShardingInfo shardingInfo;
+    final Integer shardId;
+    final DriverChannelOptions options;
+    final NodeMetricUpdater nodeMetricUpdater;
+    ProtocolVersion currentVersion;
+    boolean isNegotiating;
+    final List<ProtocolVersion> attemptedVersions;
+    final CompletableFuture<DriverChannel> resultFuture;
+
+    ConnectRequest(
+        EndPoint endPoint,
+        NodeShardingInfo shardingInfo,
+        Integer shardId,
+        DriverChannelOptions options,
+        NodeMetricUpdater nodeMetricUpdater,
+        ProtocolVersion currentVersion,
+        boolean isNegotiating,
+        List<ProtocolVersion> attemptedVersions,
+        CompletableFuture<DriverChannel> resultFuture) {
+      this.endPoint = endPoint;
+      this.shardingInfo = shardingInfo;
+      this.shardId = shardId;
+      this.options = options;
+      this.nodeMetricUpdater = nodeMetricUpdater;
+      this.currentVersion = currentVersion;
+      this.isNegotiating = isNegotiating;
+      this.attemptedVersions = attemptedVersions;
+      this.resultFuture = resultFuture;
+    }
+  }
 
-    SocketAddress resolvedAddress;
+  /**
+   * Entry point for an actual connection attempt. Resolves the endpoint address — expanding
+   * unresolved hostnames to all known IP addresses so that non-responsive individual IPs can be
+   * skipped — then delegates iteration to {@link #tryNextAddress}.
+   */
+  private void connect(ConnectRequest request) {
+    SocketAddress raw;
     try {
-      resolvedAddress = endPoint.resolve();
+      raw = request.endPoint.resolve();
     } catch (Exception e) {
-      resultFuture.completeExceptionally(e);
+      request.resultFuture.completeExceptionally(e);
       return;
     }
 
-    NettyOptions nettyOptions = context.getNettyOptions();
+    List<InetSocketAddress> candidates;
+    if (raw instanceof InetSocketAddress) {
+      InetSocketAddress inetAddr = (InetSocketAddress) raw;
+      if (inetAddr.isUnresolved()) {
+        // Hostname has not been resolved yet — expand it to all known IPs now so that we can
+        // fall back to subsequent addresses if the first one is non-responsive.
+        try {
+          InetAddress[] all = InetAddress.getAllByName(inetAddr.getHostString());
+          candidates = new ArrayList<>(all.length);
+          for (InetAddress addr : all) {
+            candidates.add(new InetSocketAddress(addr, inetAddr.getPort()));
+          }
+        } catch (UnknownHostException e) {
+          request.resultFuture.completeExceptionally(e);
+          return;
+        }
+      } else {
+        candidates = Collections.singletonList(inetAddr);
+      }
+    } else {
+      // Non-inet address (e.g. Unix domain socket) — pass through as-is.
+      candidates = Collections.singletonList(new InetSocketAddress(raw.toString(), 0));
+      tryNextAddressRaw(request, raw);
+      return;
+    }
+
+    tryNextAddress(request, candidates, 0);
+  }
+
+  /**
+   * Iterates through the candidate addresses, calling {@link #connectToAddress} for each. If an
+   * address fails for a reason other than protocol negotiation, the next candidate is tried. Only
+   * when all candidates are exhausted is the overall {@code resultFuture} failed.
+   */
+  private void tryNextAddress(
+      ConnectRequest request, List<InetSocketAddress> candidates, int index) {
+    InetSocketAddress address = candidates.get(index);
+    connectToAddress(request, address)
+        .whenComplete(
+            (channel, error) -> {
+              if (error == null) {
+                // connectToAddress already completed resultFuture on success.
+              } else if (index + 1 < candidates.size()) {
+                LOG.debug(
+                    "[{}] Failed to connect to {} ({}), trying next address",
+                    logPrefix,
+                    address,
+                    error.getMessage());
+                tryNextAddress(request, candidates, index + 1);
+              } else {
+                // Note: might be completed already if the failure happened in initializer()
+                request.resultFuture.completeExceptionally(error);
+              }
+            });
+  }
 
+  /**
+   * Performs a Netty bootstrap connect to a single, already-resolved {@link InetSocketAddress}.
+   * Handles protocol version negotiation (downgrade retries) internally, staying on the same
+   * address. Returns a {@link CompletableFuture} that completes with the {@link DriverChannel} on
+   * success, or fails with the connection error so the caller can decide whether to try a different
+   * address.
+   */
+  private CompletableFuture<DriverChannel> connectToAddress(
+      ConnectRequest request, InetSocketAddress address) {
+    CompletableFuture<DriverChannel> addressFuture = new CompletableFuture<>();
+
+    NettyOptions nettyOptions = context.getNettyOptions();
     Bootstrap bootstrap =
         new Bootstrap()
             .group(nettyOptions.ioEventLoopGroup())
             .channel(nettyOptions.channelClass())
             .option(ChannelOption.ALLOCATOR, nettyOptions.allocator())
             .handler(
-                initializer(endPoint, currentVersion, options, nodeMetricUpdater, resultFuture));
-
+                initializer(
+                    request.endPoint,
+                    request.currentVersion,
+                    request.options,
+                    request.nodeMetricUpdater,
+                    request.resultFuture));
     nettyOptions.afterBootstrapInitialized(bootstrap);
 
     ChannelFuture connectFuture;
-    if (shardId == null || shardingInfo == null) {
-      if (shardId != null) {
+    if (request.shardId == null || request.shardingInfo == null) {
+      if (request.shardId != null) {
         LOG.debug(
             "Requested connection to shard {} but shardingInfo is currently missing for Node at endpoint {}. Falling back to arbitrary local port.",
-            shardId,
-            endPoint);
+            request.shardId,
+            request.endPoint);
       }
-      connectFuture = bootstrap.connect(resolvedAddress);
+      connectFuture = bootstrap.connect(address);
     } else {
       int localPort =
-          PortAllocator.getNextAvailablePort(shardingInfo.getShardsCount(), shardId, context);
+          PortAllocator.getNextAvailablePort(
+              request.shardingInfo.getShardsCount(), request.shardId, context);
       if (localPort == -1) {
         LOG.warn(
             "Could not find free port for shard {} at {}. Falling back to arbitrary local port.",
-            shardId,
-            endPoint);
-        connectFuture = bootstrap.connect(resolvedAddress);
+            request.shardId,
+            request.endPoint);
+        connectFuture = bootstrap.connect(address);
       } else {
-        connectFuture = bootstrap.connect(resolvedAddress, new InetSocketAddress(localPort));
+        connectFuture = bootstrap.connect(address, new InetSocketAddress(localPort));
       }
     }
 
@@ -267,11 +375,12 @@ private void connect(
           if (connectFuture.isSuccess()) {
             Channel channel = connectFuture.channel();
             DriverChannel driverChannel =
-                new DriverChannel(endPoint, channel, context.getWriteCoalescer(), currentVersion);
+                new DriverChannel(
+                    request.endPoint, channel, context.getWriteCoalescer(), request.currentVersion);
             // If this is the first successful connection, remember the protocol version and
             // cluster name for future connections.
-            if (isNegotiating) {
-              ChannelFactory.this.protocolVersion = currentVersion;
+            if (request.isNegotiating) {
+              ChannelFactory.this.protocolVersion = request.currentVersion;
             }
             if (ChannelFactory.this.clusterName == null) {
               ChannelFactory.this.clusterName = driverChannel.getClusterName();
@@ -294,39 +403,81 @@ private void connect(
                             ConsistencyLevel.LOCAL_QUORUM.name()));
               }
             }
-            resultFuture.complete(driverChannel);
+            request.resultFuture.complete(driverChannel);
+            addressFuture.complete(driverChannel);
           } else {
             Throwable error = connectFuture.cause();
-            if (error instanceof UnsupportedProtocolVersionException && isNegotiating) {
-              attemptedVersions.add(currentVersion);
+            if (error instanceof UnsupportedProtocolVersionException && request.isNegotiating) {
+              // Protocol negotiation failure: downgrade and retry the SAME address.
+              request.attemptedVersions.add(request.currentVersion);
               Optional<ProtocolVersion> downgraded =
-                  context.getProtocolVersionRegistry().downgrade(currentVersion);
+                  context.getProtocolVersionRegistry().downgrade(request.currentVersion);
               if (downgraded.isPresent()) {
                 LOG.debug(
                     "[{}] Failed to connect with protocol {}, retrying with {}",
                     logPrefix,
-                    currentVersion,
+                    request.currentVersion,
                     downgraded.get());
-                connect(
-                    endPoint,
-                    shardingInfo,
-                    shardId,
-                    options,
-                    nodeMetricUpdater,
-                    downgraded.get(),
-                    true,
-                    attemptedVersions,
-                    resultFuture);
+                request.currentVersion = downgraded.get();
+                connectToAddress(request, address)
+                    .whenComplete(
+                        (ch, err) -> {
+                          if (err != null) addressFuture.completeExceptionally(err);
+                          else addressFuture.complete(ch);
+                        });
               } else {
-                resultFuture.completeExceptionally(
+                Throwable negotiationError =
                     UnsupportedProtocolVersionException.forNegotiation(
-                        endPoint, attemptedVersions));
+                        request.endPoint, request.attemptedVersions);
+                request.resultFuture.completeExceptionally(negotiationError);
+                addressFuture.completeExceptionally(negotiationError);
               }
             } else {
-              // Note: might be completed already if the failure happened in initializer(), this is
-              // fine
-              resultFuture.completeExceptionally(error);
+              addressFuture.completeExceptionally(error);
+            }
+          }
+        });
+
+    return addressFuture;
+  }
+
+  /**
+   * Handles the non-{@link InetSocketAddress} path (e.g. Unix domain sockets) by connecting
+   * directly to the raw address without multi-address fallback.
+   */
+  private void tryNextAddressRaw(ConnectRequest request, SocketAddress rawAddress) {
+    NettyOptions nettyOptions = context.getNettyOptions();
+    Bootstrap bootstrap =
+        new Bootstrap()
+            .group(nettyOptions.ioEventLoopGroup())
+            .channel(nettyOptions.channelClass())
+            .option(ChannelOption.ALLOCATOR, nettyOptions.allocator())
+            .handler(
+                initializer(
+                    request.endPoint,
+                    request.currentVersion,
+                    request.options,
+                    request.nodeMetricUpdater,
+                    request.resultFuture));
+    nettyOptions.afterBootstrapInitialized(bootstrap);
+
+    ChannelFuture connectFuture = bootstrap.connect(rawAddress);
+    connectFuture.addListener(
+        cf -> {
+          if (connectFuture.isSuccess()) {
+            Channel channel = connectFuture.channel();
+            DriverChannel driverChannel =
+                new DriverChannel(
+                    request.endPoint, channel, context.getWriteCoalescer(), request.currentVersion);
+            if (request.isNegotiating) {
+              ChannelFactory.this.protocolVersion = request.currentVersion;
             }
+            if (ChannelFactory.this.clusterName == null) {
+              ChannelFactory.this.clusterName = driverChannel.getClusterName();
+            }
+            request.resultFuture.complete(driverChannel);
+          } else {
+            request.resultFuture.completeExceptionally(connectFuture.cause());
           }
         });
   }

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

@@ -201,6 +201,48 @@ 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>With {@code RESOLVE_CONTACT_POINTS=false} (the default), the hostname is kept unresolved
+   * until connection time. Previously, only the first IP returned by DNS was tried, causing an
+   * {@code AllNodesFailedException} when that IP was unreachable. The fix expands the hostname to
+   * all IPs at connection time and tries each one in sequence.
+   */
+  @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.RESOLVE_CONTACT_POINTS.getRawOption(), false)
+              .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() {