feat: add EndPoint.resolveAll() for multi-address DNS expansion (DRIVER-201)

Addresses the endpoint-API aspect of DRIVER-201.

Problem: EndPoint.resolve() returns a single SocketAddress. When a
hostname maps to multiple IPs, the driver can only try the first one
and fails with AllNodesFailedException if it is unreachable — the
remaining IPs are invisible to the connection layer.

Solution (per @dkropachev's architectural direction):
- Deprecate EndPoint.resolve(). Add EndPoint.resolveAll() with a default
  implementation that wraps resolve() in a single-element array for
  backward compatibility with third-party implementations.
- DefaultEndPoint.resolveAll(): if the stored InetSocketAddress is
  unresolved, calls InetAddress.getAllByName() to expand the hostname
  to all known IPs, returning one InetSocketAddress per IP. Falls back
  to the single-element unresolved address if DNS fails, so the connect
  attempt surfaces a descriptive error rather than returning empty.
- SniEndPoint.resolveAll(): re-resolves the proxy hostname on each call
  and returns all A-records sorted by IP, enabling the caller to try
  each proxy address in sequence.
- ClientRoutesEndPoint.resolveAll(): delegates to resolve() (single-
  address topology-monitor lookup) and wraps in a one-element array.
- ChannelFactory.connect(): replaced endPoint.resolve() with
  endPoint.resolveAll(). Iterates through the returned candidates via
  tryNextCandidate(); on per-address failure logs and tries the next;
  only fails the overall resultFuture when all candidates are exhausted.
  Protocol-version negotiation (downgrade retries) is scoped to the
  same address via connectToAddress(), which is semantically correct.

Tests:
- DefaultEndPointTest: 3 new cases — already-resolved passthrough,
  unresolved hostname expansion, unresolvable hostname fallback.
- SniEndPointTest: new class with cases for resolveAll() happy path,
  unresolvable host exception, and resolve() sanity check.
- All 13 existing ChannelFactory tests continue to pass (LocalEndPoint
  uses the default single-element resolveAll() via the interface default).

Author: nikagra

core/src/main/java/com/datastax/dse/driver/api/core/auth/DseGssApiAuthProviderBase.java

@@ -291,6 +291,8 @@ protected static class GssApiAuthenticator extends BaseDseAuthenticator {
     private SaslClient saslClient;
     private EndPoint endPoint;
 
+    @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll();
+    // Kerberos authentication needs a single canonical hostname for SASL service name resolution.
     protected GssApiAuthenticator(
         GssApiOptions options, EndPoint endPoint, String serverAuthenticator) {
       super(serverAuthenticator);

core/src/main/java/com/datastax/dse/driver/internal/core/insights/InsightsClient.java

@@ -288,6 +288,8 @@ private InsightsStatusData createStatusData() {
         .build();
   }
 
+  @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll();
+  // address reporting needs a single canonical address per node.
   private Map<String, SessionStateForNode> getConnectedNodes() {
     Map<Node, ChannelPool> pools = driverContext.getPoolManager().getPools();
     return pools.entrySet().stream()
@@ -302,6 +304,8 @@ private SessionStateForNode constructSessionStateForNode(Map.Entry<Node, Channel
         entry.getKey().getOpenConnections(), entry.getValue().getInFlight());
   }
 
+  @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll();
+  // address reporting needs a single canonical address per node.
   private InsightsStartupData createStartupData() {
     Map<String, String> startupOptions = driverContext.getStartupOptions();
     return InsightsStartupData.builder()
@@ -454,6 +458,8 @@ private PoolSizeByHostDistance getPoolSizeByHostDistance() {
         0);
   }
 
+  @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll();
+  // address reporting needs a single canonical address for the control connection.
   private String getControlConnectionSocketAddress() {
     SocketAddress controlConnectionAddress = controlConnection.channel().getEndPoint().resolve();
     return AddressFormatter.nullSafeToString(controlConnectionAddress);

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

@@ -18,28 +18,58 @@
 package com.datastax.oss.driver.api.core.metadata;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
-import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 
 /**
  * Encapsulates the information needed to open connections to a node.
  *
  * <p>By default, the driver assumes plain TCP connections, and this is just a wrapper around an
- * {@link InetSocketAddress}. However, more complex deployment scenarios might use a custom
+ * {@link java.net.InetSocketAddress}. However, more complex deployment scenarios might use a custom
  * implementation that contains additional information; for example, if the nodes are accessed
  * through a proxy with SNI routing, an SNI server name is needed in addition to the proxy address.
  */
 public interface EndPoint {
 
   /**
-   * Resolves this instance to a socket address.
+   * Resolves this instance to a single socket address.
    *
    * <p>This will be called each time the driver opens a new connection to the node. The returned
    * address cannot be null.
+   *
+   * @deprecated Use {@link #resolveAll()} instead. When a hostname maps to multiple IPs (e.g. in
+   *     dynamic DNS environments) only one address is returned here, causing the driver to miss
+   *     fallback IPs when the first one is unreachable. {@code resolveAll()} returns the full set.
    */
+  @Deprecated
   @NonNull
   SocketAddress resolve();
 
+  /**
+   * Resolves this instance to all known socket addresses.
+   *
+   * <p>This is called each time the driver opens a new connection to the node. For endpoints backed
+   * by a plain IP address the array contains exactly one element. For endpoints whose hostname
+   * resolves to multiple IPs (e.g. a DNS round-robin entry) all addresses are returned so that the
+   * driver can try each one in sequence and fall back gracefully when individual IPs are
+   * unreachable.
+   *
+   * <p>The default implementation wraps {@link #resolve()} and returns a single-element array.
+   * Implementations that can supply multiple addresses should override this method.
+   *
+   * <p>The returned array must not be null and must contain at least one element.
+   *
+   * <p><b>Timeout note:</b> {@link com.datastax.oss.driver.internal.core.channel.ChannelFactory}
+   * tries each address in sequence. If a hostname resolves to N addresses and each attempt times
+   * out, the worst-case connection time before declaring a node unreachable is {@code N ×
+   * advanced.connection.connect-timeout}. In practice DNS round-robin entries have only a small
+   * number of records, so this is rarely a concern, but callers should be aware of this when
+   * configuring connect timeouts.
+   */
+  @NonNull
+  default SocketAddress[] resolveAll() {
+    return new SocketAddress[] {resolve()};
+  }
+
   /**
    * Returns an alternate string representation for use in node-level metric names.
    *

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

@@ -219,14 +219,119 @@ private void connect(
       List<ProtocolVersion> attemptedVersions,
       CompletableFuture<DriverChannel> resultFuture) {
 
-    SocketAddress resolvedAddress;
+    SocketAddress[] candidates;
     try {
-      resolvedAddress = endPoint.resolve();
+      candidates = endPoint.resolveAll();
+      if (candidates == null || candidates.length == 0) {
+        resultFuture.completeExceptionally(
+            new IllegalArgumentException(
+                "EndPoint.resolveAll() must return a non-null, non-empty array: " + endPoint));
+        return;
+      }
     } catch (Exception e) {
       resultFuture.completeExceptionally(e);
       return;
     }
 
+    tryNextCandidate(
+        endPoint,
+        shardingInfo,
+        shardId,
+        options,
+        nodeMetricUpdater,
+        currentVersion,
+        isNegotiating,
+        attemptedVersions,
+        resultFuture,
+        candidates,
+        0);
+  }
+
+  /**
+   * Iterates through the candidate addresses from {@link EndPoint#resolveAll()}. Tries each one in
+   * sequence; if an address fails for a reason other than protocol-version negotiation exhaustion,
+   * the next candidate is tried. Only when all candidates are exhausted is the overall {@code
+   * resultFuture} failed.
+   *
+   * <p><b>Timeout note:</b> addresses are tried serially, so the worst-case time before failure is
+   * {@code N × connect-timeout} where N is the number of candidates. This is an intentional
+   * tradeoff: failing immediately on the first unreachable IP would prevent fallback to healthy
+   * ones. In practice DNS entries have only a small number of records.
+   */
+  private void tryNextCandidate(
+      EndPoint endPoint,
+      NodeShardingInfo shardingInfo,
+      Integer shardId,
+      DriverChannelOptions options,
+      NodeMetricUpdater nodeMetricUpdater,
+      ProtocolVersion currentVersion,
+      boolean isNegotiating,
+      List<ProtocolVersion> attemptedVersions,
+      CompletableFuture<DriverChannel> resultFuture,
+      SocketAddress[] candidates,
+      int index) {
+
+    SocketAddress candidate = candidates[index];
+    CompletableFuture<DriverChannel> perAddressFuture = new CompletableFuture<>();
+    connectToAddress(
+        endPoint,
+        shardingInfo,
+        shardId,
+        options,
+        nodeMetricUpdater,
+        currentVersion,
+        isNegotiating,
+        attemptedVersions,
+        perAddressFuture,
+        candidate);
+
+    perAddressFuture.whenComplete(
+        (channel, error) -> {
+          if (error == null) {
+            resultFuture.complete(channel);
+          } else if (index + 1 < candidates.length) {
+            LOG.debug(
+                "[{}] Failed to connect to {} ({}), trying next address",
+                logPrefix,
+                candidate,
+                error.getMessage());
+            tryNextCandidate(
+                endPoint,
+                shardingInfo,
+                shardId,
+                options,
+                nodeMetricUpdater,
+                currentVersion,
+                isNegotiating,
+                attemptedVersions,
+                resultFuture,
+                candidates,
+                index + 1);
+          } else {
+            // Note: might be completed already if the failure happened in initializer()
+            resultFuture.completeExceptionally(error);
+          }
+        });
+  }
+
+  /**
+   * Performs a Netty bootstrap connect to a single, already-resolved address. Handles
+   * protocol-version negotiation (downgrade retries) internally, staying on the same address. Uses
+   * {@code perAddressFuture} so {@link #tryNextCandidate} can distinguish a per-address TCP failure
+   * (try the next IP) from a successful protocol handshake.
+   */
+  private void connectToAddress(
+      EndPoint endPoint,
+      NodeShardingInfo shardingInfo,
+      Integer shardId,
+      DriverChannelOptions options,
+      NodeMetricUpdater nodeMetricUpdater,
+      ProtocolVersion currentVersion,
+      boolean isNegotiating,
+      List<ProtocolVersion> attemptedVersions,
+      CompletableFuture<DriverChannel> perAddressFuture,
+      SocketAddress resolvedAddress) {
+
     NettyOptions nettyOptions = context.getNettyOptions();
 
     Bootstrap bootstrap =
@@ -235,7 +340,8 @@ private void connect(
             .channel(nettyOptions.channelClass())
             .option(ChannelOption.ALLOCATOR, nettyOptions.allocator())
             .handler(
-                initializer(endPoint, currentVersion, options, nodeMetricUpdater, resultFuture));
+                initializer(
+                    endPoint, currentVersion, options, nodeMetricUpdater, perAddressFuture));
 
     nettyOptions.afterBootstrapInitialized(bootstrap);
 
@@ -294,7 +400,7 @@ private void connect(
                             ConsistencyLevel.LOCAL_QUORUM.name()));
               }
             }
-            resultFuture.complete(driverChannel);
+            perAddressFuture.complete(driverChannel);
           } else {
             Throwable error = connectFuture.cause();
             if (error instanceof UnsupportedProtocolVersionException && isNegotiating) {
@@ -307,7 +413,8 @@ private void connect(
                     logPrefix,
                     currentVersion,
                     downgraded.get());
-                connect(
+                // Stay on the same address for protocol-version downgrade retries.
+                connectToAddress(
                     endPoint,
                     shardingInfo,
                     shardId,
@@ -316,16 +423,17 @@ private void connect(
                     downgraded.get(),
                     true,
                     attemptedVersions,
-                    resultFuture);
+                    perAddressFuture,
+                    resolvedAddress);
               } else {
-                resultFuture.completeExceptionally(
+                perAddressFuture.completeExceptionally(
                     UnsupportedProtocolVersionException.forNegotiation(
                         endPoint, attemptedVersions));
               }
             } else {
               // Note: might be completed already if the failure happened in initializer(), this is
               // fine
-              resultFuture.completeExceptionally(error);
+              perAddressFuture.completeExceptionally(error);
             }
           }
         });

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

@@ -76,6 +76,20 @@ public SocketAddress resolve() {
     return fallbackEndPoint.resolve();
   }
 
+  /**
+   * Returns all socket addresses for this endpoint.
+   *
+   * <p>Delegates to {@link #resolve()} to obtain the single address provided by the topology
+   * monitor (or the fallback endpoint), then returns it as a one-element array. The topology
+   * monitor resolves each node to exactly one address by design (via a per-host-id lookup), so
+   * multi-address expansion is not applicable here.
+   */
+  @NonNull
+  @Override
+  public SocketAddress[] resolveAll() {
+    return new SocketAddress[] {resolve()};
+  }
+
   @Override
   public boolean equals(Object other) {
     if (other == this) {

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

@@ -20,7 +20,10 @@
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.Serializable;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.UnknownHostException;
 import java.util.Objects;
 
 public class DefaultEndPoint implements EndPoint, Serializable {
@@ -41,6 +44,47 @@ public InetSocketAddress resolve() {
     return address;
   }
 
+  /**
+   * Returns all socket addresses for this endpoint.
+   *
+   * <p>If the stored address is unresolved (i.e. the driver was configured with {@code
+   * RESOLVE_CONTACT_POINTS=false} and the hostname has not been looked up yet), this method calls
+   * {@link InetAddress#getAllByName(String)} to expand the hostname to every IP it resolves to.
+   * Each resolved IP is returned as an {@link InetSocketAddress} with the same port as the
+   * original. If the hostname resolves to only one IP, or if the address is already resolved, a
+   * single-element array is returned.
+   *
+   * <p>If DNS resolution fails, falls back to a single-element array containing {@link #resolve()}.
+   *
+   * <p><b>Note on resolver:</b> DNS lookup is performed via {@link
+   * InetAddress#getAllByName(String)} on the calling thread, bypassing any custom Netty {@code
+   * AddressResolverGroup} configured via {@link
+   * com.datastax.oss.driver.api.core.config.DefaultDriverOption#NETTY_ADMIN_SIZE}. This is
+   * consistent with how {@link SniEndPoint} and {@link
+   * com.datastax.oss.driver.internal.core.metadata.MetadataManager#getResolvedContactPoints()}
+   * perform DNS resolution elsewhere in the driver. Users who rely on a custom Netty resolver
+   * should supply pre-resolved {@link java.net.InetSocketAddress} instances instead of hostnames.
+   */
+  @NonNull
+  @Override
+  public SocketAddress[] resolveAll() {
+    if (!address.isUnresolved()) {
+      return new SocketAddress[] {address};
+    }
+    try {
+      InetAddress[] all = InetAddress.getAllByName(address.getHostString());
+      SocketAddress[] result = new SocketAddress[all.length];
+      for (int i = 0; i < all.length; i++) {
+        result[i] = new InetSocketAddress(all[i], address.getPort());
+      }
+      return result;
+    } catch (UnknownHostException e) {
+      // Fallback: return the single unresolved address; the connect attempt will fail with a
+      // descriptive error rather than silently returning an empty array.
+      return new SocketAddress[] {address};
+    }
+  }
+
   @Override
   public boolean equals(Object other) {
     if (other == this) {

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

@@ -701,6 +701,8 @@ private Optional<NodeInfo> findInPeers(
   // Current versions of Cassandra (3.11 at the time of writing), require the same port for all
   // nodes. As a consequence, the port is not stored in system tables.
   // We save it the first time we get a control connection channel.
+  @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); a single
+  // canonical address is all that is needed here to extract the port.
   protected void savePort(DriverChannel channel) {
     if (port < 0) {
       SocketAddress address = channel.getEndPoint().resolve();
@@ -723,6 +725,8 @@ protected void savePort(DriverChannel channel) {
    *     otherwise.
    */
   @Nullable
+  @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); a single
+  // canonical address is all that is needed here for the peer-vs-local comparison.
   protected InetSocketAddress getBroadcastRpcAddress(
       @NonNull AdminRow row, @NonNull EndPoint localEndPoint) {