client routes: disable fallback to broadcast address when route is missing

When client routes are configured, the driver must connect exclusively
through proxies (addresses from system.client_routes). The previous
behavior silently fell back to the node's direct broadcast address when
no route entry existed for a host_id, which caused races and silent
misbehaviour upon adding new nodes (time window between node addition
and client route entry publication).

Now ClientRoutesEndPoint.resolve() throws IllegalStateException when
no route is found, causing the node to go DOWN and enter the standard
reconnection loop until a CLIENT_ROUTES_CHANGE event populates the
route. This matches the Rust driver's approach of failing address
translation on missing entries.

Changes:
- ClientRoutesEndPoint: remove fallbackEndPoint field; resolve() throws
  with WARN log instead of falling back to broadcast address
- ClientRoutesTopologyMonitor.buildNodeEndPoint(): remove fallback
  computation; throw on null host_id instead of delegating to super
- Tests updated to verify exception-throwing behavior

Author: nikagra

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

@@ -27,33 +27,30 @@
 import java.net.SocketAddress;
 import java.util.Objects;
 import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class ClientRoutesEndPoint implements EndPoint {
+  private static final Logger LOG = LoggerFactory.getLogger(ClientRoutesEndPoint.class);
+
   private final UUID hostId;
   private final ClientRoutesTopologyMonitor topologyMonitor;
   private final String metricPrefix;
-  @NonNull private final EndPoint fallbackEndPoint;
 
   /**
    * @param topologyMonitor the topology monitor used to resolve the endpoint address on demand.
    * @param hostId the host UUID identifying this node in the cluster.
    * @param broadcastInetAddress the node's broadcast address (from system.peers or system.local),
    *     used to build a stable metric prefix. May be {@code null} if the address could not be
    *     determined, in which case the hostId is used as the metric prefix instead.
-   * @param fallbackEndPoint the default endpoint to fall back to when {@code
-   *     topologyMonitor.resolve()} returns {@code null}, i.e. when this node is not accessed via a
-   *     cloud private endpoint. Must not be {@code null}.
    */
   public ClientRoutesEndPoint(
       @NonNull ClientRoutesTopologyMonitor topologyMonitor,
       @NonNull UUID hostId,
-      @Nullable InetAddress broadcastInetAddress,
-      @NonNull EndPoint fallbackEndPoint) {
+      @Nullable InetAddress broadcastInetAddress) {
     this.topologyMonitor =
         Objects.requireNonNull(topologyMonitor, "Topology monitor cannot be null");
     this.hostId = Objects.requireNonNull(hostId, "HOST uuid cannot be null");
-    this.fallbackEndPoint =
-        Objects.requireNonNull(fallbackEndPoint, "Fallback endpoint cannot be null");
     this.metricPrefix = buildMetricPrefix(broadcastInetAddress, hostId);
   }
 
@@ -73,7 +70,20 @@ public SocketAddress resolve() {
     } catch (IOException e) {
       throw new UncheckedIOException("DNS resolution failed for host_id=" + hostId, e);
     }
-    return fallbackEndPoint.resolve();
+    // When client routes are configured, the driver must connect exclusively through proxies.
+    // Falling back to the node's broadcast address would bypass the proxy infrastructure and
+    // cause silent misbehaviour (e.g. during the window between adding a new node and posting
+    // its client route entry). The node will remain DOWN and the reconnection loop will retry
+    // until a CLIENT_ROUTES_CHANGE event populates the route.
+    LOG.warn(
+        "No client route entry found for host_id={}. "
+            + "The node will remain DOWN until a route is published via CLIENT_ROUTES_CHANGE.",
+        hostId);
+    throw new IllegalStateException(
+        "No client route entry found for host_id="
+            + hostId
+            + ". Will not connect to the node's broadcast address because client routes "
+            + "are configured; the driver must connect exclusively through proxies.");
   }
 
   @Override

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

@@ -461,12 +461,15 @@ protected EndPoint buildNodeEndPoint(
       LOG.warn(
           "[{}] host_id is null in system row for address {} — cannot assign a client route. "
               + "This may indicate corrupted system tables. "
-              + "Falling back to default endpoint resolution.",
+              + "The node will be ignored because client routes are configured "
+              + "and the driver must connect exclusively through proxies.",
           logPrefix,
           broadcastRpcAddress);
-      return super.buildNodeEndPoint(row, broadcastRpcAddress, localEndPoint);
+      throw new IllegalStateException(
+          "host_id is null in system row for address "
+              + broadcastRpcAddress
+              + "; cannot build a ClientRoutesEndPoint without a host_id");
     }
-    EndPoint fallback = super.buildNodeEndPoint(row, broadcastRpcAddress, localEndPoint);
     InetAddress broadcastInetAddress = null;
     if (broadcastRpcAddress != null) {
       broadcastInetAddress = broadcastRpcAddress.getAddress();
@@ -477,7 +480,7 @@ protected EndPoint buildNodeEndPoint(
     if (broadcastInetAddress == null) {
       broadcastInetAddress = row.getInetAddress("peer");
     }
-    return new ClientRoutesEndPoint(this, hostId, broadcastInetAddress, fallback);
+    return new ClientRoutesEndPoint(this, hostId, broadcastInetAddress);
   }
 
   /**

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

@@ -21,7 +21,6 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.when;
 
-import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import java.io.UncheckedIOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -36,7 +35,6 @@
 public class ClientRoutesEndPointTest {
 
   @Mock private ClientRoutesTopologyMonitor topologyMonitor;
-  @Mock private EndPoint fallbackEndPoint;
 
   // ---- resolve() ----------------------------------------------------------
 
@@ -46,32 +44,31 @@ public void should_resolve_via_topology_monitor() throws UnknownHostException {
     InetSocketAddress expected = new InetSocketAddress("127.0.0.1", 9042);
     when(topologyMonitor.resolve(hostId)).thenReturn(expected);
 
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
     assertThat(ep.resolve()).isEqualTo(expected);
   }
 
   @Test
-  public void should_fallback_when_resolve_returns_null() throws UnknownHostException {
+  public void should_throw_when_resolve_returns_null() throws UnknownHostException {
     UUID hostId = UUID.randomUUID();
-    InetSocketAddress fallbackAddr = new InetSocketAddress("10.0.0.1", 9042);
     when(topologyMonitor.resolve(hostId)).thenReturn(null);
-    when(fallbackEndPoint.resolve()).thenReturn(fallbackAddr);
 
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
-    assertThat(ep.resolve()).isEqualTo(fallbackAddr);
+    assertThatThrownBy(ep::resolve)
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("No client route entry found")
+        .hasMessageContaining(hostId.toString())
+        .hasMessageContaining("exclusively through proxies");
   }
 
   @Test
   public void should_wrap_io_exceptions_in_unchecked_io_exception() throws UnknownHostException {
     UUID hostId = UUID.randomUUID();
     when(topologyMonitor.resolve(hostId)).thenThrow(new UnknownHostException("no-such-host"));
 
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
     assertThatThrownBy(ep::resolve)
         .isInstanceOf(UncheckedIOException.class)
@@ -85,8 +82,7 @@ public void should_reflect_route_changes_on_subsequent_resolve() throws UnknownH
     InetSocketAddress addr2 = new InetSocketAddress("10.0.0.1", 9043);
     when(topologyMonitor.resolve(hostId)).thenReturn(addr1);
 
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
     assertThat(ep.resolve()).isEqualTo(addr1);
 
@@ -101,30 +97,25 @@ public void should_reflect_route_changes_on_subsequent_resolve() throws UnknownH
   @Test
   public void should_be_equal_when_same_host_id() {
     UUID hostId = UUID.randomUUID();
-    ClientRoutesEndPoint ep1 =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
-    ClientRoutesEndPoint ep2 =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep1 = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
+    ClientRoutesEndPoint ep2 = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
     assertThat(ep1).isEqualTo(ep2);
     assertThat(ep1.hashCode()).isEqualTo(ep2.hashCode());
   }
 
   @Test
   public void should_not_be_equal_when_different_host_id() {
-    ClientRoutesEndPoint ep1 =
-        new ClientRoutesEndPoint(topologyMonitor, UUID.randomUUID(), null, fallbackEndPoint);
-    ClientRoutesEndPoint ep2 =
-        new ClientRoutesEndPoint(topologyMonitor, UUID.randomUUID(), null, fallbackEndPoint);
+    ClientRoutesEndPoint ep1 = new ClientRoutesEndPoint(topologyMonitor, UUID.randomUUID(), null);
+    ClientRoutesEndPoint ep2 = new ClientRoutesEndPoint(topologyMonitor, UUID.randomUUID(), null);
 
     assertThat(ep1).isNotEqualTo(ep2);
   }
 
   @Test
   public void should_not_be_equal_to_non_client_routes_endpoint() {
     UUID hostId = UUID.randomUUID();
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
     assertThat(ep).isNotEqualTo("not an endpoint");
     assertThat(ep).isNotEqualTo(null);
@@ -135,8 +126,7 @@ public void should_not_be_equal_to_non_client_routes_endpoint() {
   @Test
   public void should_use_host_id_as_metric_prefix_when_address_is_null() {
     UUID hostId = UUID.randomUUID();
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
     assertThat(ep.asMetricPrefix()).isEqualTo(hostId.toString());
   }
@@ -145,8 +135,7 @@ public void should_use_host_id_as_metric_prefix_when_address_is_null() {
   public void should_format_ipv4_metric_prefix() throws Exception {
     UUID hostId = UUID.randomUUID();
     InetAddress ipv4 = InetAddress.getByAddress(new byte[] {10, 0, 0, 1});
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, ipv4, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, ipv4);
 
     assertThat(ep.asMetricPrefix()).isEqualTo("10_0_0_1_" + hostId);
   }
@@ -156,8 +145,7 @@ public void should_format_ipv6_metric_prefix() throws Exception {
     UUID hostId = UUID.randomUUID();
     InetAddress ipv6 =
         InetAddress.getByAddress(new byte[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1});
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, ipv6, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, ipv6);
 
     // IPv6 keeps colons (consistent with DefaultEndPoint), dots replaced by underscores
     assertThat(ep.asMetricPrefix()).isEqualTo("0:0:0:0:0:0:0:1_" + hostId);
@@ -168,8 +156,7 @@ public void should_format_ipv6_metric_prefix() throws Exception {
   @Test
   public void should_return_host_id_as_string() {
     UUID hostId = UUID.randomUUID();
-    ClientRoutesEndPoint ep =
-        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
 
     assertThat(ep.toString()).isEqualTo("ClientRoutesEndPoint(" + hostId + ")");
   }

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

@@ -38,7 +38,6 @@
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.net.InetSocketAddress;
-import java.net.SocketAddress;
 import java.net.UnknownHostException;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -1041,19 +1040,16 @@ public void should_not_propagate_exception_when_query_fails() throws Exception {
   // ---- buildNodeEndPoint fallback -----------------------------------------
 
   @Test
-  public void should_build_default_endpoint_when_host_id_is_null() {
-    // row.getUuid("host_id") returns null, triggering the hostId == null
-    // branch in buildNodeEndPoint which delegates to super.buildNodeEndPoint().
+  public void should_throw_when_host_id_is_null() {
+    // row.getUuid("host_id") returns null — with client routes configured, the driver
+    // must not fall back to direct broadcast address, so buildNodeEndPoint throws.
     AdminRow row = Mockito.mock(AdminRow.class);
     when(row.getUuid("host_id")).thenReturn(null);
-    when(row.contains("peer")).thenReturn(false); // local-node row → super returns localEndPoint
     EndPoint localEndPoint = Mockito.mock(EndPoint.class);
 
-    EndPoint result = handler.buildNodeEndPoint(row, null, localEndPoint);
-
-    // hostId == null branch → super.buildNodeEndPoint() is called → returns localEndPoint
-    assertThat(result).isNotInstanceOf(ClientRoutesEndPoint.class);
-    assertThat(result).isSameAs(localEndPoint);
+    assertThatThrownBy(() -> handler.buildNodeEndPoint(row, null, localEndPoint))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("host_id is null");
   }
 
   @Test
@@ -1065,7 +1061,6 @@ public void should_build_client_routes_endpoint_when_host_id_non_null() {
     UUID hostId = UUID.randomUUID();
     AdminRow row = Mockito.mock(AdminRow.class);
     when(row.getUuid("host_id")).thenReturn(hostId);
-    when(row.contains("peer")).thenReturn(false);
     EndPoint localEndPoint = Mockito.mock(EndPoint.class);
 
     EndPoint result = handler.buildNodeEndPoint(row, null, localEndPoint);
@@ -1138,24 +1133,23 @@ hostId1, new ClientRouteRecord(hostId1, "127.0.0.1", 9042),
   }
 
   @Test
-  public void should_resolve_to_fallback_when_no_route_for_host_id() {
-    // Simulates a node that is not accessed via PrivateLink (no route in cache for its host_id).
-    // resolve() must return the regular endpoint address (the fallback), not throw.
+  public void should_throw_when_no_route_for_host_id() {
+    // Simulates a node that has no route in the client_routes table (e.g. newly added node
+    // before route is published). resolve() must throw instead of falling back to broadcast
+    // address, to prevent the driver from bypassing proxy infrastructure.
     UUID hostId = UUID.randomUUID();
-    InetSocketAddress fallbackAddress = new InetSocketAddress("127.0.0.99", 9999);
     AdminRow row = Mockito.mock(AdminRow.class);
     when(row.getUuid("host_id")).thenReturn(hostId);
-    when(row.contains("peer")).thenReturn(false);
     EndPoint localEndPoint = Mockito.mock(EndPoint.class);
-    when(localEndPoint.resolve()).thenReturn(fallbackAddress);
 
     EndPoint endpoint = handler.buildNodeEndPoint(row, null, localEndPoint);
     assertThat(endpoint).isInstanceOf(ClientRoutesEndPoint.class);
 
-    // Cache is empty (no PrivateLink route) → resolves to the regular endpoint address
-    SocketAddress resolved = ((ClientRoutesEndPoint) endpoint).resolve();
-    assertThat(resolved).isEqualTo(fallbackAddress);
-    Mockito.verify(localEndPoint).resolve();
+    // Cache is empty (no route for this host_id) → must throw, not fall back
+    assertThatThrownBy(endpoint::resolve)
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("No client route entry found")
+        .hasMessageContaining(hostId.toString());
   }
 
   // ---- savePort() --------------------------------------------------------