client routes: add exclusive-proxy config flag for strict proxy enforcement

Introduces advanced.client-routes.exclusive-proxy (boolean, default false)
to control whether the driver falls back to a node's broadcast address when
no route entry exists in system.client_routes.

When false (default, backward-compatible): nodes without a route entry are
reached directly via their broadcast address, preserving support for mixed
proxy/direct topologies.

When true: any node without a route entry is treated as unreachable; the
driver throws IllegalStateException with a WARN log and the node stays
DOWN until CLIENT_ROUTES_CHANGE populates the route. This matches the
behavior previously hardcoded in the prior commit.

Changes:
- DefaultDriverOption: add CLIENT_ROUTES_EXCLUSIVE_PROXY
- TypedDriverOption: add typed Boolean constant
- reference.conf: add exclusive-proxy = false with full documentation
- ClientRoutesEndPoint: restore fallbackEndPoint field; resolve() gates
  the throw/fallback on the exclusiveProxy flag
- ClientRoutesTopologyMonitor: read config flag; pass fallback endpoint
  (null when exclusive-proxy=true) and flag to ClientRoutesEndPoint
- Tests: update constructor calls; rename and add test cases for both
  modes; add createHandlerWithExclusiveProxy() helper

Author: nikagra

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

@@ -1119,7 +1119,22 @@ public enum DefaultDriverOption implements DriverOption {
    *
    * <p>Value type: boolean
    */
-  CLIENT_ROUTES_SHARD_AWARENESS_ENABLED("advanced.client-routes.shard-awarness-enabled");
+  CLIENT_ROUTES_SHARD_AWARENESS_ENABLED("advanced.client-routes.shard-awarness-enabled"),
+
+  /**
+   * Whether the driver connects exclusively through proxies when client routes are configured.
+   *
+   * <p>When {@code true}, any node whose {@code host_id} does not appear in the {@code
+   * system.client_routes} table is treated as unreachable; the driver will never fall back to the
+   * node's broadcast address. The node stays DOWN and the reconnection loop retries until a {@code
+   * CLIENT_ROUTES_CHANGE} event populates the route.
+   *
+   * <p>When {@code false} (the default), nodes that have no route entry are contacted directly
+   * using their broadcast address, preserving backward-compatible mixed proxy/direct topologies.
+   *
+   * <p>Value type: boolean
+   */
+  CLIENT_ROUTES_EXCLUSIVE_PROXY("advanced.client-routes.exclusive-proxy");
 
   private final String path;
 

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

@@ -954,6 +954,15 @@ public String toString() {
       new TypedDriverOption<>(
           DefaultDriverOption.CLIENT_ROUTES_SHARD_AWARENESS_ENABLED, GenericType.BOOLEAN);
 
+  /**
+   * Whether the driver connects exclusively through proxies when client routes are configured. When
+   * {@code false} (default), nodes without a route entry are contacted directly via their broadcast
+   * address.
+   */
+  public static final TypedDriverOption<Boolean> CLIENT_ROUTES_EXCLUSIVE_PROXY =
+      new TypedDriverOption<>(
+          DefaultDriverOption.CLIENT_ROUTES_EXCLUSIVE_PROXY, GenericType.BOOLEAN);
+
   private static Iterable<TypedDriverOption<?>> introspectBuiltInValues() {
     try {
       ImmutableList.Builder<TypedDriverOption<?>> result = ImmutableList.builder();

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

@@ -36,21 +36,33 @@ public class ClientRoutesEndPoint implements EndPoint {
   private final UUID hostId;
   private final ClientRoutesTopologyMonitor topologyMonitor;
   private final String metricPrefix;
+  @Nullable private final EndPoint fallbackEndPoint;
+  private final boolean exclusiveProxy;
 
   /**
    * @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. Ignored (and may be {@code null}) when {@code exclusiveProxy} is
+   *     {@code true}.
+   * @param exclusiveProxy when {@code true}, {@link #resolve()} throws instead of falling back to
+   *     {@code fallbackEndPoint} if no route is found.
    */
   public ClientRoutesEndPoint(
       @NonNull ClientRoutesTopologyMonitor topologyMonitor,
       @NonNull UUID hostId,
-      @Nullable InetAddress broadcastInetAddress) {
+      @Nullable InetAddress broadcastInetAddress,
+      @Nullable EndPoint fallbackEndPoint,
+      boolean exclusiveProxy) {
     this.topologyMonitor =
         Objects.requireNonNull(topologyMonitor, "Topology monitor cannot be null");
     this.hostId = Objects.requireNonNull(hostId, "HOST uuid cannot be null");
+    this.fallbackEndPoint = fallbackEndPoint;
+    this.exclusiveProxy = exclusiveProxy;
     this.metricPrefix = buildMetricPrefix(broadcastInetAddress, hostId);
   }
 
@@ -70,11 +82,17 @@ public SocketAddress resolve() {
     } catch (IOException e) {
       throw new UncheckedIOException("DNS resolution failed for host_id=" + hostId, e);
     }
-    // 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.
+    if (!exclusiveProxy && fallbackEndPoint != null) {
+      // Default (backward-compatible) mode: fall back to the node's broadcast address.
+      // This supports mixed proxy/direct topologies where some nodes are behind the private
+      // endpoint and others are reached directly.
+      return fallbackEndPoint.resolve();
+    }
+    // Exclusive-proxy mode: the driver must connect only 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.",

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

@@ -82,6 +82,7 @@ public class ClientRoutesTopologyMonitor extends DefaultTopologyMonitor {
   private final String logPrefix;
   private final AtomicReference<Map<UUID, ClientRouteRecord>> resolvedRoutesCache;
   private final boolean useSSL;
+  private final boolean exclusiveProxy;
   private volatile boolean closed = false;
   private final AtomicInteger consecutiveEmptyResults = new AtomicInteger(0);
 
@@ -147,6 +148,11 @@ public ClientRoutesTopologyMonitor(
     this.logPrefix = context.getSessionName();
     this.resolvedRoutesCache = new AtomicReference<>(Collections.emptyMap());
     this.useSSL = context.getSslEngineFactory().isPresent();
+    this.exclusiveProxy =
+        context
+            .getConfig()
+            .getDefaultProfile()
+            .getBoolean(DefaultDriverOption.CLIENT_ROUTES_EXCLUSIVE_PROXY);
   }
 
   @Override
@@ -480,7 +486,9 @@ protected EndPoint buildNodeEndPoint(
     if (broadcastInetAddress == null) {
       broadcastInetAddress = row.getInetAddress("peer");
     }
-    return new ClientRoutesEndPoint(this, hostId, broadcastInetAddress);
+    EndPoint fallback =
+        exclusiveProxy ? null : super.buildNodeEndPoint(row, broadcastRpcAddress, localEndPoint);
+    return new ClientRoutesEndPoint(this, hostId, broadcastInetAddress, fallback, exclusiveProxy);
   }
 
   /**

core/src/main/resources/reference.conf

@@ -1173,6 +1173,19 @@ datastax-java-driver {
     # Default: false
     shard-awarness-enabled = false
 
+    # When true, the driver connects exclusively through proxies and will never fall back to a
+    # node's broadcast address when no route entry exists in system.client_routes. Any node
+    # without a route is treated as unreachable; it goes DOWN and the reconnection loop retries
+    # until a CLIENT_ROUTES_CHANGE event populates the route.
+    #
+    # When false (default), nodes that have no route entry are reached directly using their
+    # broadcast address, preserving backward-compatible mixed proxy/direct topologies where some
+    # nodes are behind the private endpoint and others are not.
+    #
+    # Required: no
+    # Default: false
+    exclusive-proxy = false
+
   }
 
   # Whether to resolve the addresses passed to `basic.contact-points`.

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

@@ -21,6 +21,7 @@
 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;
@@ -35,6 +36,7 @@
 public class ClientRoutesEndPointTest {
 
   @Mock private ClientRoutesTopologyMonitor topologyMonitor;
+  @Mock private EndPoint fallbackEndPoint;
 
   // ---- resolve() ----------------------------------------------------------
 
@@ -44,17 +46,18 @@ 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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
 
     assertThat(ep.resolve()).isEqualTo(expected);
   }
 
   @Test
-  public void should_throw_when_resolve_returns_null() throws UnknownHostException {
+  public void should_throw_when_exclusive_proxy_and_resolve_returns_null()
+      throws UnknownHostException {
     UUID hostId = UUID.randomUUID();
     when(topologyMonitor.resolve(hostId)).thenReturn(null);
 
-    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, true);
 
     assertThatThrownBy(ep::resolve)
         .isInstanceOf(IllegalStateException.class)
@@ -63,12 +66,26 @@ public void should_throw_when_resolve_returns_null() throws UnknownHostException
         .hasMessageContaining("exclusively through proxies");
   }
 
+  @Test
+  public void should_fall_back_to_broadcast_when_exclusive_proxy_disabled()
+      throws UnknownHostException {
+    UUID hostId = UUID.randomUUID();
+    InetSocketAddress fallbackAddress = new InetSocketAddress("10.0.0.1", 9042);
+    when(topologyMonitor.resolve(hostId)).thenReturn(null);
+    when(fallbackEndPoint.resolve()).thenReturn(fallbackAddress);
+
+    ClientRoutesEndPoint ep =
+        new ClientRoutesEndPoint(topologyMonitor, hostId, null, fallbackEndPoint, false);
+
+    assertThat(ep.resolve()).isEqualTo(fallbackAddress);
+  }
+
   @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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
 
     assertThatThrownBy(ep::resolve)
         .isInstanceOf(UncheckedIOException.class)
@@ -82,7 +99,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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
 
     assertThat(ep.resolve()).isEqualTo(addr1);
 
@@ -97,25 +114,27 @@ 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);
-    ClientRoutesEndPoint ep2 = new ClientRoutesEndPoint(topologyMonitor, hostId, null);
+    ClientRoutesEndPoint ep1 = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
+    ClientRoutesEndPoint ep2 = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
 
     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);
-    ClientRoutesEndPoint ep2 = new ClientRoutesEndPoint(topologyMonitor, UUID.randomUUID(), null);
+    ClientRoutesEndPoint ep1 =
+        new ClientRoutesEndPoint(topologyMonitor, UUID.randomUUID(), null, null, false);
+    ClientRoutesEndPoint ep2 =
+        new ClientRoutesEndPoint(topologyMonitor, UUID.randomUUID(), null, null, false);
 
     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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
 
     assertThat(ep).isNotEqualTo("not an endpoint");
     assertThat(ep).isNotEqualTo(null);
@@ -126,7 +145,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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
 
     assertThat(ep.asMetricPrefix()).isEqualTo(hostId.toString());
   }
@@ -135,7 +154,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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, ipv4, null, false);
 
     assertThat(ep.asMetricPrefix()).isEqualTo("10_0_0_1_" + hostId);
   }
@@ -145,7 +164,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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, ipv6, null, false);
 
     // IPv6 keeps colons (consistent with DefaultEndPoint), dots replaced by underscores
     assertThat(ep.asMetricPrefix()).isEqualTo("0:0:0:0:0:0:0:1_" + hostId);
@@ -156,7 +175,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);
+    ClientRoutesEndPoint ep = new ClientRoutesEndPoint(topologyMonitor, hostId, null, null, false);
 
     assertThat(ep.toString()).isEqualTo("ClientRoutesEndPoint(" + hostId + ")");
   }

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

@@ -142,6 +142,8 @@ public void setup() {
     when(defaultProfile.getDuration(DefaultDriverOption.CONTROL_CONNECTION_TIMEOUT))
         .thenReturn(Duration.ofSeconds(5));
     when(defaultProfile.getBoolean(DefaultDriverOption.RECONNECT_ON_INIT)).thenReturn(false);
+    when(defaultProfile.getBoolean(DefaultDriverOption.CLIENT_ROUTES_EXCLUSIVE_PROXY))
+        .thenReturn(false);
     when(context.getSslEngineFactory()).thenReturn(Optional.empty());
     ClientRoutesConfig config =
         ClientRoutesConfig.builder()
@@ -162,6 +164,21 @@ private void initHandler() {
     handler.init();
   }
 
+  /**
+   * Creates a fresh handler with the given {@code exclusiveProxy} setting, sharing all other
+   * context stubs from {@link #setup()}.
+   */
+  private TestableClientRoutesTopologyMonitor createHandlerWithExclusiveProxy(
+      boolean exclusiveProxy) {
+    when(defaultProfile.getBoolean(DefaultDriverOption.CLIENT_ROUTES_EXCLUSIVE_PROXY))
+        .thenReturn(exclusiveProxy);
+    ClientRoutesConfig config =
+        ClientRoutesConfig.builder()
+            .addEndpoint(new ClientRouteProxy(connectionId, "host1"))
+            .build();
+    return new TestableClientRoutesTopologyMonitor(context, config);
+  }
+
   // ---- resolve() -------------------------------------------------------
 
   @Test
@@ -1133,16 +1150,16 @@ hostId1, new ClientRouteRecord(hostId1, "127.0.0.1", 9042),
   }
 
   @Test
-  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
+  public void should_throw_when_no_route_for_host_id_and_exclusive_proxy_enabled() {
+    // With exclusive-proxy=true: resolve() must throw instead of falling back to broadcast
     // address, to prevent the driver from bypassing proxy infrastructure.
+    TestableClientRoutesTopologyMonitor exclusiveHandler = createHandlerWithExclusiveProxy(true);
     UUID hostId = UUID.randomUUID();
     AdminRow row = Mockito.mock(AdminRow.class);
     when(row.getUuid("host_id")).thenReturn(hostId);
     EndPoint localEndPoint = Mockito.mock(EndPoint.class);
 
-    EndPoint endpoint = handler.buildNodeEndPoint(row, null, localEndPoint);
+    EndPoint endpoint = exclusiveHandler.buildNodeEndPoint(row, null, localEndPoint);
     assertThat(endpoint).isInstanceOf(ClientRoutesEndPoint.class);
 
     // Cache is empty (no route for this host_id) → must throw, not fall back
@@ -1152,6 +1169,26 @@ public void should_throw_when_no_route_for_host_id() {
         .hasMessageContaining(hostId.toString());
   }
 
+  @Test
+  public void should_fall_back_to_broadcast_when_no_route_and_exclusive_proxy_disabled()
+      throws Exception {
+    // With exclusive-proxy=false (default): resolve() delegates to the fallback endpoint when
+    // no route exists, supporting mixed proxy/direct topologies.
+    UUID hostId = UUID.randomUUID();
+    AdminRow row = Mockito.mock(AdminRow.class);
+    when(row.getUuid("host_id")).thenReturn(hostId);
+    EndPoint localEndPoint = Mockito.mock(EndPoint.class);
+    InetSocketAddress directAddress = new InetSocketAddress("10.0.0.1", 9042);
+    when(localEndPoint.resolve()).thenReturn(directAddress);
+
+    // handler uses exclusiveProxy=false (set in setup())
+    EndPoint endpoint = handler.buildNodeEndPoint(row, null, localEndPoint);
+    assertThat(endpoint).isInstanceOf(ClientRoutesEndPoint.class);
+
+    // Cache is empty → falls back to the direct broadcast address
+    assertThat(endpoint.resolve()).isEqualTo(directAddress);
+  }
+
   // ---- savePort() --------------------------------------------------------
 
   @Test