feat: fall back to contact points on control reconnect (DRIVER-201)

Default fallback-to-original-contact-points to true: once a control
reconnection round has exhausted the live nodes, the unresolved contact
points are appended, so Netty resolves their names again and a cluster
that moved is found. Every metadata node, the control node included,
otherwise holds an address resolved once. Monitors that re-resolve on
their own opt out through TopologyMonitor.reresolvesNodeAddresses(),
unless the live-node plan is empty. The control node's identity is
applied to the channel at adoption, so every node refresh reads one
endpoint. HeartbeatIT turns the fallback off; MockResolverIT gains the
recovery case.

Fixes #215

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: nikagra

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

@@ -701,8 +701,21 @@ public enum DefaultDriverOption implements DriverOption {
   CONTROL_CONNECTION_AGREEMENT_WARN("advanced.control-connection.schema-agreement.warn-on-failure"),
 
   /**
-   * Whether to forcibly add original contact points held by MetadataManager to the reconnection
-   * plan, in case there is no live nodes available according to LBP. Experimental.
+   * Whether to append the original contact points held by MetadataManager to the control
+   * connection's reconnection plan, after the live nodes reported by the load balancing policy.
+   * Defaults to {@code true}.
+   *
+   * <p>This is the driver's DNS re-resolution path. A metadata node holds an address resolved once
+   * and never re-resolved, the node the control connection reached included; a contact point given
+   * as a hostname is kept unresolved (the default) and looked up again through Netty's resolver on
+   * each connect, so once the live nodes are exhausted the contact points find a cluster that moved
+   * to new addresses, as soon as the JVM's DNS cache ({@code networkaddress.cache.ttl}) has
+   * expired. A resolved contact point ({@code advanced.resolve-contact-points = true}, or a
+   * programmatic resolved address) is appended as it is and never re-resolved.
+   *
+   * <p>Skipped when the topology monitor re-resolves node addresses on its own ({@code
+   * TopologyMonitor#reresolvesNodeAddresses()}: the Cloud SNI proxy, and client routes with full
+   * route coverage), unless the live-node plan is empty.
    *
    * <p>Value-type: boolean
    */

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

@@ -369,7 +369,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) {
     map.put(TypedDriverOption.CONTROL_CONNECTION_AGREEMENT_INTERVAL, Duration.ofMillis(200));
     map.put(TypedDriverOption.CONTROL_CONNECTION_AGREEMENT_TIMEOUT, Duration.ofSeconds(10));
     map.put(TypedDriverOption.CONTROL_CONNECTION_AGREEMENT_WARN, true);
-    map.put(TypedDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, false);
+    map.put(TypedDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, true);
     map.put(TypedDriverOption.PREPARE_ON_ALL_NODES, true);
     map.put(TypedDriverOption.REPREPARE_ENABLED, true);
     map.put(TypedDriverOption.REPREPARE_CHECK_SYSTEM_TABLE, false);

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

@@ -600,7 +600,15 @@ public String toString() {
   public static final TypedDriverOption<Boolean> CONTROL_CONNECTION_AGREEMENT_WARN =
       new TypedDriverOption<>(
           DefaultDriverOption.CONTROL_CONNECTION_AGREEMENT_WARN, GenericType.BOOLEAN);
-  /** Whether to forcibly try original contacts if no live nodes are available */
+  /**
+   * Whether to append the original contact points to the control connection's reconnection plan,
+   * after the live nodes reported by the load balancing policy (defaults to {@code true}).
+   *
+   * <p>A contact point given as a hostname is kept unresolved and looked up again on each connect,
+   * so this is how the driver picks up changed DNS records once the live nodes are exhausted. It is
+   * skipped for a topology monitor that re-resolves node addresses on its own, unless the live-node
+   * plan is empty.
+   */
   public static final TypedDriverOption<Boolean> CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS =
       new TypedDriverOption<>(
           DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, GenericType.BOOLEAN);

core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java

@@ -61,7 +61,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Objects;
 import java.util.Queue;
 import java.util.WeakHashMap;
 import java.util.concurrent.CompletableFuture;
@@ -459,6 +458,19 @@ private void connect(
                       connect(nodes, errors, onSuccess, onFailure);
                     } else {
                       LOG.debug("[{}] New channel opened {}", logPrefix, channel);
+                      // Identify the node before the channel becomes visible to anything else.
+                      // Every read of channel.getEndPoint() from here on -- the topology monitor's
+                      // three node-refresh methods included -- must see the same endpoint: if one
+                      // refresh identified the control node by the address it was dialled at and
+                      // the next by the address it answered on, the second would compare the two
+                      // (a blocking lookup, for a mixed resolved/unresolved pair) and then rewrite
+                      // the node's endpoint and clear its metrics.
+                      EndPoint identity =
+                          context.getTopologyMonitor().connectedNodeEndPoint(channel);
+                      if (identity != null && identity != channel.getEndPoint()) {
+                        channel.setEndPoint(identity);
+                        LOG.debug("[{}] Control channel identified as {}", logPrefix, identity);
+                      }
                       DriverChannel previousChannel = ControlConnection.this.channel;
                       ControlConnection.this.channel = channel;
                       controlNodeState = new ControlNodeState(null, node);
@@ -499,6 +511,42 @@ private void connect(
                                           node,
                                           new Exception("Channel closed during endpoint resolve")));
                                   connect(nodes, newErrors, onSuccess, onFailure);
+                                } else if (isUnusableForControl(resolvedNode)) {
+                                  // A contact point's identity is only known once the resolve
+                                  // above completes, and events name the metadata node rather than
+                                  // the placeholder we dialled, so isControlNode() cannot match the
+                                  // two -- and comparing their endpoints there would mean the
+                                  // blocking lookup this path exists to avoid. So re-run the
+                                  // pre-connect check against the node we actually identified.
+                                  // lastNodeDistance/lastNodeState are cumulative, so this also
+                                  // catches an event that arrived before the channel opened, which
+                                  // the check on the placeholder never saw.
+                                  controlNodeState = ControlNodeState.NONE;
+                                  LOG.debug(
+                                      "[{}] New channel opened ({}) but {} is ignored or forced "
+                                          + "down, closing and trying next node",
+                                      logPrefix,
+                                      channel,
+                                      resolvedNode);
+                                  // Null out before forceClose() so that onChannelClosed() does not
+                                  // start a redundant reconnection on top of the connect() retry
+                                  // below.
+                                  ControlConnection.this.channel = null;
+                                  channel.forceClose();
+                                  // Recorded like every other reason a candidate was dropped: a
+                                  // round in which all of them are dropped here would otherwise
+                                  // fail with a causeless NoNodeAvailableException, saying nothing
+                                  // about the channels that were opened and deliberately closed.
+                                  List<Entry<Node, Throwable>> newErrors =
+                                      (errors == null) ? new ArrayList<>() : errors;
+                                  newErrors.add(
+                                      new SimpleEntry<>(
+                                          node,
+                                          new Exception(
+                                              "Control node "
+                                                  + resolvedNode
+                                                  + " is ignored or forced down")));
+                                  connect(nodes, newErrors, onSuccess, onFailure);
                                 } else {
                                   controlNodeState = new ControlNodeState(resolvedNode, null);
                                   context
@@ -545,13 +593,13 @@ private CompletionStage<Node> resolveChannelNodeIfNeeded(
           .thenComposeAsync(
               nodeInfo -> {
                 EndPoint resolvedEp = nodeInfo.getEndPoint();
-                // Compared by reference, not equals(): DefaultEndPoint.equals resolves the
-                // unresolved side of a mixed comparison, a blocking lookup on this admin executor,
-                // and it would answer "equal" for the very case this exists for (a hostname contact
-                // point identified by the address it reached), skipping the adoption. Adopting the
+                // A monitor that builds its endpoints from the identity row rather than from the
+                // socket (the Cloud SNI proxy, client routes) only knows the node's endpoint now;
+                // connectedNodeEndPoint() left the channel's alone for those. Compared by
+                // reference, not equals(): DefaultEndPoint.equals resolves the unresolved side of
+                // a mixed comparison, a blocking lookup on this admin executor. Adopting the
                 // monitor's instance also makes the channel and the metadata node share it, so
-                // every
-                // later comparison of the two short-circuits on identity.
+                // every later comparison of the two short-circuits on identity.
                 if (resolvedEp != null && resolvedEp != channel.getEndPoint()) {
                   channel.setEndPoint(resolvedEp);
                   LOG.debug("[{}] Control channel endpoint upgraded to {}", logPrefix, resolvedEp);
@@ -694,12 +742,47 @@ private boolean isControlNode(Node eventNode) {
           && eventNode.getHostId().equals(state.current.getHostId())) {
         return true;
       }
-      if (state.current == null
-          && state.pending != null
-          && Objects.equals(eventNode.getEndPoint(), state.pending.getEndPoint())) {
-        return true;
-      }
-      return false;
+      // Reference identity, not endpoint equality: the reconnection plan now appends the
+      // unresolved contact-point nodes (fallback-to-original-contact-points defaults to true), so a
+      // mixed resolved/unresolved pair is routine here, and DefaultEndPoint.equals resolves the
+      // unresolved side of one -- a blocking lookup on this admin executor, during the very DNS
+      // outage the fallback exists for. Nodes carry identity semantics (no equals override, and
+      // lastNodeState/lastNodeDistance already key on the instance) and connect() stores the plan's
+      // own node as pending, so an event about that node carries the same object.
+      //
+      // An event naming the metadata node for the host we are still identifying cannot be matched
+      // here at all -- pending is the placeholder, and the two are different instances by
+      // construction (MetadataManager.registerNode never reuses a contact-point node). Endpoint
+      // equality appeared to cover that case, but only by paying the blocking lookup above.
+      // connect() instead re-checks the identified node with isUnusableForControl() once the
+      // resolve completes, which closes the window without any DNS.
+      return state.current == null && state.pending != null && eventNode == state.pending;
+    }
+
+    /**
+     * Whether an event has already marked this node unusable for the control connection: the load
+     * balancing policy ignored it, or it was removed or forced down.
+     *
+     * <p>Mirrors the two checks {@link #connect} runs before it adopts a freshly opened channel.
+     * Those stay separate rather than calling this: they answer from records read before the
+     * channel was opened, this one from the records as they stand now, and each keeps its own
+     * diagnostic message.
+     *
+     * <p>Answers only for a node some event has named. Both records are keyed by node instance (and
+     * weakly), so this sees a node the driver already knows -- either a query-plan node, or a host
+     * id {@code MetadataManager#registerNode} found in the metadata and handed back. It cannot see
+     * a node registered fresh from a contact point: {@code registerNode} mints a new instance for
+     * an unknown host id and re-inserts it into the metadata before this runs, so a host the driver
+     * had removed is resurrected rather than rejected. Detecting that needs a removal record keyed
+     * by host id, with a defined lifetime, which this check deliberately does not attempt:
+     * rejecting every host id absent from the metadata would also reject the genuinely new nodes of
+     * a cluster that moved, which is the case the fallback exists to recover.
+     */
+    private boolean isUnusableForControl(Node node) {
+      NodeState state = lastNodeState.get(node);
+      return lastNodeDistance.get(node) == NodeDistance.IGNORED
+          || (lastNodeState.containsKey(node)
+              && (state == null /*(removed)*/ || state == NodeState.FORCED_DOWN));
     }
 
     private void onDistanceEvent(DistanceEvent event) {

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

@@ -21,12 +21,14 @@
 import com.datastax.oss.driver.api.core.config.ClientRoutesConfig;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminRequestHandler;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminResult;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminRow;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import com.datastax.oss.driver.internal.core.clientroutes.ClientRouteRecord;
 import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
+import com.datastax.oss.driver.internal.core.util.AddressUtils;
 import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
@@ -480,6 +482,45 @@ protected EndPoint buildNodeEndPoint(
     return new ClientRoutesEndPoint(this, hostId, broadcastInetAddress, fallback);
   }
 
+  @Override
+  public boolean reresolvesNodeAddresses() {
+    // A ClientRoutesEndPoint looks its route hostname up on every connect, but only while a route
+    // exists for that host id; without one it falls back to a static, already-resolved address. So
+    // this answers true only while every known node has a live route, and "every" has to mean at
+    // least one: an empty node set (before the first refresh, or after every node was removed) must
+    // not suppress the contact-point fallback at the one moment it is the only way back. A closed
+    // monitor re-resolves nothing at all, however complete its cache still looks.
+    if (closed) {
+      return false;
+    }
+    // Routes first, nodes second, and the order is the safety property rather than a style
+    // choice: the two are published independently, by components that do not share this thread,
+    // so a pass necessarily mixes one snapshot with a possibly newer other. Read this way, a node
+    // added since the routes were published is missing from them and reports false; read the other
+    // way it would be missing from the node set instead, and its absence of a route would go
+    // unnoticed -- claiming a re-resolution that node does not have.
+    Map<UUID, ClientRouteRecord> routes = resolvedRoutesCache.get();
+    // getNodes() is already keyed by host id, so iterating its keys is exactly "every known node",
+    // and the node objects themselves are never asked anything.
+    Map<UUID, Node> nodes = context.getMetadataManager().getMetadata().getNodes();
+    if (nodes.isEmpty()) {
+      return false;
+    }
+    for (UUID hostId : nodes.keySet()) {
+      ClientRouteRecord route = routes.get(hostId);
+      // A route reaches DNS only if its address holds a name: resolve() calls
+      // InetAddress.getByName(), which hands an IP literal straight back, so a literal-valued
+      // route re-resolves nothing and must not suppress the contact-point fallback -- that
+      // fallback is the only way such a node ever learns a new address. The address column takes
+      // either form and nothing validates which (scylladb/java-driver#1064), so decide it here,
+      // per record and lexically, without a lookup.
+      if (route == null || AddressUtils.isIpLiteral(route.getHostname())) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   /**
    * Builds the CQL query to fetch client routes.
    *

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

@@ -44,4 +44,12 @@ protected EndPoint buildNodeEndPoint(
     UUID hostId = Objects.requireNonNull(row.getUuid("host_id"));
     return new SniEndPoint(cloudProxyAddress, hostId.toString());
   }
+
+  @Override
+  public boolean reresolvesNodeAddresses() {
+    // Every node is reached through the SNI proxy, whose hostname SniEndPoint#resolve() looks up on
+    // every call, so addresses stay current on their own: appending the contact points as a DNS
+    // fallback would add nothing, and could resurrect nodes this monitor has removed.
+    return true;
+  }
 }

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

@@ -351,7 +351,7 @@ public CompletionStage<NodeInfo> getChannelNodeInfo(DriverChannel channel) {
     if (closeFuture.isDone()) {
       return CompletableFutures.failedFuture(new IllegalStateException("closed"));
     }
-    EndPoint localEndPoint = connectedNodeEndPoint(channel);
+    EndPoint localEndPoint = channel.getEndPoint();
     return query(channel, buildQuery(localColumns, "system.local", "key='local'"))
         .thenApply(
             result -> {
@@ -626,8 +626,8 @@ protected DefaultNodeInfo.Builder nodeInfoBuilder(
   }
 
   /**
-   * The endpoint to register the node at the other end of {@code channel} under, when the node is
-   * identified for the first time.
+   * The endpoint to identify the node at the other end of {@code channel} by, applied to the
+   * channel by {@code ControlConnection} when it adopts it, before any query is sent.
    *
    * <p>A contact point's endpoint is an unresolved name by default, and that name is not the
    * node's: every node ever reached through the contact point would be registered under it, and
@@ -640,11 +640,14 @@ protected DefaultNodeInfo.Builder nodeInfoBuilder(
    *
    * <p>Anything else is returned as configured: a resolved endpoint ({@code resolve-contact-points
    * = true}, or a programmatic resolved address), a third-party {@link EndPoint}, or a channel
-   * whose remote address is not an {@link InetSocketAddress}. For the default deployment this moves
-   * the control node's metric name from the contact point's name to its own address, once, and pool
-   * connections opened to it later verify TLS against that address, like every peer's.
+   * whose remote address is not an {@link InetSocketAddress}. Those already name a node, so there
+   * is nothing to derive -- a resolved contact point keeps identifying the control node by the name
+   * it was configured with, deliberately. For the default deployment this moves the control node's
+   * metric name from the contact point's name to its own address, once, and pool connections opened
+   * to it later verify TLS against that address, like every peer's.
    */
-  private static EndPoint connectedNodeEndPoint(DriverChannel channel) {
+  @Override
+  public EndPoint connectedNodeEndPoint(DriverChannel channel) {
     EndPoint configured = channel.getEndPoint();
     if (!(configured instanceof DefaultEndPoint)
         || !((DefaultEndPoint) configured).resolve().isUnresolved()) {