fix: use native transport port for broadcastRpcAddress in ClientRoutes

ClientRoutesTopologyMonitor.savePort() was saving the NLB proxy port
(e.g. 29043) as the fallback for broadcastRpcAddress construction.
This caused Metadata.findNode() to fail matching TOPOLOGY_CHANGE
REMOVED_NODE events (which carry the real native transport port 9042),
so decommissioned nodes were never removed from metadata.

Add a nativeTransportPort field to ClientRoutesConfig (default 9042,
configurable via advanced.client-routes.native-transport-port) and set
it in the ClientRoutesTopologyMonitor constructor. The savePort()
override is now a no-op since the port is already initialized.

broadcastRpcAddress is used only for event matching — connections go
through ClientRoutesEndPoint which resolves via NLB proxy addresses
in the routes cache.

Author: dkropachev

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

@@ -63,6 +63,16 @@ public final class ClientRoutesConfig {
 
   private static final String DEFAULT_TABLE_NAME = "system.client_routes";
 
+  /**
+   * Default native transport port used by Scylla/Cassandra nodes. This is used as the fallback port
+   * for {@code broadcastRpcAddress} when the system tables do not include port information (e.g.
+   * older Scylla versions without {@code system.peers_v2}). The {@code broadcastRpcAddress} is how
+   * the driver matches TOPOLOGY_CHANGE events (node added/removed) to nodes in its metadata — it is
+   * NOT used for opening connections. Connections go through {@code ClientRoutesEndPoint} which
+   * resolves via the NLB proxy addresses in the routes cache.
+   */
+  public static final int DEFAULT_NATIVE_TRANSPORT_PORT = 9042;
+
   /**
    * Pattern for valid unquoted CQL table names: must start with a letter or underscore, followed by
    * letters, digits, or underscores. Optionally qualified with a keyspace prefix using the same
@@ -73,8 +83,10 @@ public final class ClientRoutesConfig {
 
   private final List<ClientRouteProxy> endpoints;
   private final String tableName;
+  private final int nativeTransportPort;
 
-  private ClientRoutesConfig(List<ClientRouteProxy> endpoints, String tableName) {
+  private ClientRoutesConfig(
+      List<ClientRouteProxy> endpoints, String tableName, int nativeTransportPort) {
     if (endpoints == null || endpoints.isEmpty()) {
       throw new IllegalArgumentException("At least one endpoint must be specified");
     }
@@ -86,6 +98,7 @@ private ClientRoutesConfig(List<ClientRouteProxy> endpoints, String tableName) {
     }
     this.endpoints = Collections.unmodifiableList(new ArrayList<>(endpoints));
     this.tableName = tableName;
+    this.nativeTransportPort = nativeTransportPort;
   }
 
   /**
@@ -108,6 +121,18 @@ public String getTableName() {
     return tableName;
   }
 
+  /**
+   * Returns the native transport port of the cluster nodes. This port is used as the fallback for
+   * building {@code broadcastRpcAddress} when system tables lack port columns. The {@code
+   * broadcastRpcAddress} is used only to match TOPOLOGY_CHANGE events to metadata nodes — it is NOT
+   * used to open connections (those go through the NLB proxy endpoints from the routes cache).
+   *
+   * @return the native transport port (defaults to {@value #DEFAULT_NATIVE_TRANSPORT_PORT}).
+   */
+  public int getNativeTransportPort() {
+    return nativeTransportPort;
+  }
+
   /**
    * Creates a new builder for constructing a {@link ClientRoutesConfig}.
    *
@@ -127,12 +152,14 @@ public boolean equals(Object o) {
       return false;
     }
     ClientRoutesConfig that = (ClientRoutesConfig) o;
-    return endpoints.equals(that.endpoints) && tableName.equals(that.tableName);
+    return endpoints.equals(that.endpoints)
+        && tableName.equals(that.tableName)
+        && nativeTransportPort == that.nativeTransportPort;
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(endpoints, tableName);
+    return Objects.hash(endpoints, tableName, nativeTransportPort);
   }
 
   @Override
@@ -143,6 +170,8 @@ public String toString() {
         + ", tableName='"
         + tableName
         + '\''
+        + ", nativeTransportPort="
+        + nativeTransportPort
         + '}';
   }
 
@@ -151,6 +180,7 @@ public static final class Builder {
     private final List<ClientRouteProxy> endpoints = new ArrayList<>();
     private final Set<String> seenConnectionIds = new HashSet<>();
     private String tableName = DEFAULT_TABLE_NAME;
+    private int nativeTransportPort = DEFAULT_NATIVE_TRANSPORT_PORT;
 
     /**
      * Adds an endpoint to the configuration.
@@ -208,6 +238,21 @@ Builder withTableName(@NonNull String tableName) {
       return this;
     }
 
+    /**
+     * Sets the native transport port of the cluster nodes. This is used as the fallback port for
+     * {@code broadcastRpcAddress} when system tables do not include port information. Only needed
+     * for clusters using a non-standard native transport port. Defaults to {@value
+     * #DEFAULT_NATIVE_TRANSPORT_PORT}.
+     *
+     * @param port the native transport port.
+     * @return this builder.
+     */
+    @NonNull
+    public Builder withNativeTransportPort(int port) {
+      this.nativeTransportPort = port;
+      return this;
+    }
+
     /**
      * Builds the {@link ClientRoutesConfig} with the configured endpoints and table name.
      *
@@ -216,7 +261,7 @@ Builder withTableName(@NonNull String tableName) {
      */
     @NonNull
     public ClientRoutesConfig build() {
-      return new ClientRoutesConfig(endpoints, tableName);
+      return new ClientRoutesConfig(endpoints, tableName, nativeTransportPort);
     }
   }
 }

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

@@ -1097,7 +1097,17 @@ public enum DefaultDriverOption implements DriverOption {
    *
    * <p>Value type: list of HOCON objects
    */
-  CLIENT_ROUTES_ENDPOINTS("advanced.client-routes.endpoints");
+  CLIENT_ROUTES_ENDPOINTS("advanced.client-routes.endpoints"),
+
+  /**
+   * The native transport port of the cluster nodes.
+   *
+   * <p>Used as the fallback port for {@code broadcastRpcAddress} when system tables lack port
+   * columns. Only needed for clusters using a non-standard native transport port. Defaults to 9042.
+   *
+   * <p>Value type: {@link Integer}
+   */
+  CLIENT_ROUTES_NATIVE_TRANSPORT_PORT("advanced.client-routes.native-transport-port");
 
   private final String path;
 

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

@@ -398,6 +398,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) {
         "PRESERVE_REPLICA_ORDER");
     // CLIENT_ROUTES_ENDPOINTS is intentionally omitted: it is a list-of-objects (compound HOCON
     // values) with no sensible scalar default, analogous to how CONFIG_RELOAD_INTERVAL is omitted.
+    map.put(TypedDriverOption.CLIENT_ROUTES_NATIVE_TRANSPORT_PORT, 9042);
   }
 
   @Immutable

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

@@ -944,6 +944,11 @@ public String toString() {
   // DriverExecutionProfile API); it is excluded from the TypedDriverOptionTest consistency check
   // accordingly.
 
+  /** The native transport port of the cluster nodes, used for broadcastRpcAddress fallback. */
+  public static final TypedDriverOption<Integer> CLIENT_ROUTES_NATIVE_TRANSPORT_PORT =
+      new TypedDriverOption<>(
+          DefaultDriverOption.CLIENT_ROUTES_NATIVE_TRANSPORT_PORT, GenericType.INTEGER);
+
   private static Iterable<TypedDriverOption<?>> introspectBuiltInValues() {
     try {
       ImmutableList.Builder<TypedDriverOption<?>> result = ImmutableList.builder();

core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java

@@ -545,6 +545,11 @@ ClientRoutesConfig buildClientRoutesConfigFromFile() {
       builder.addEndpoint(new ClientRouteProxy(connectionId, connectionAddr));
     }
 
+    if (defaultProfile.isDefined(DefaultDriverOption.CLIENT_ROUTES_NATIVE_TRANSPORT_PORT)) {
+      builder.withNativeTransportPort(
+          defaultProfile.getInt(DefaultDriverOption.CLIENT_ROUTES_NATIVE_TRANSPORT_PORT));
+    }
+
     return builder.build();
   }
 

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

@@ -130,6 +130,7 @@ public ClientRoutesTopologyMonitor(
       @NonNull InternalDriverContext context, @NonNull ClientRoutesConfig config) {
     super(context);
     this.config = config;
+    this.port = config.getNativeTransportPort();
     this.configuredConnectionIds =
         Collections.unmodifiableList(
             config.getEndpoints().stream()
@@ -438,36 +439,15 @@ private void onClientRoutesUpdateEvent(ClientRoutesUpdateEvent event) {
   }
 
   /**
-   * Overrides the default port discovery to use the port from the client routes cache instead of
-   * the control connection channel. When connecting through an NLB proxy, the channel port is the
-   * proxy port, not the real server port. The {@code system.client_routes} table has the correct
-   * port.
-   *
-   * <p>All nodes in a Scylla/Cassandra cluster use the same native transport port, so any route's
-   * port is correct. We use the minimum host_id for deterministic selection.
+   * No-op: the port is set in the constructor from {@link ClientRoutesConfig#getNativeTransportPort
+   * ()}. The default implementation would save the control channel's endpoint port, which is the
+   * NLB proxy port — not the real native transport port. Using the proxy port would cause {@link
+   * #getBroadcastRpcAddress} to build incorrect {@code broadcastRpcAddress} values, preventing
+   * {@code Metadata.findNode()} from matching TOPOLOGY_CHANGE events (which carry the real native
+   * transport port).
    */
   @Override
-  protected void savePort(DriverChannel channel) {
-    if (port < 0) {
-      Map<UUID, ClientRouteRecord> routes = resolvedRoutesCache.get();
-      if (!routes.isEmpty()) {
-        // Pick the route with the smallest host_id for deterministic behavior.
-        UUID minId = null;
-        ClientRouteRecord chosen = null;
-        for (Map.Entry<UUID, ClientRouteRecord> entry : routes.entrySet()) {
-          if (minId == null || entry.getKey().compareTo(minId) < 0) {
-            minId = entry.getKey();
-            chosen = entry.getValue();
-          }
-        }
-        if (chosen != null && chosen.getPort() > 0) {
-          port = chosen.getPort();
-          return;
-        }
-      }
-    }
-    super.savePort(channel);
-  }
+  protected void savePort(DriverChannel channel) {}
 
   @NonNull
   @Override

core/src/main/resources/reference.conf

@@ -1146,6 +1146,17 @@ datastax-java-driver {
     # Required: yes (if client routes are to be used)
     # endpoints = []
 
+    # The native transport port of the cluster nodes. Used as the fallback port for
+    # broadcastRpcAddress when system tables lack port columns. The broadcastRpcAddress is used
+    # only to match TOPOLOGY_CHANGE events to metadata nodes — it is NOT used for opening
+    # connections (those go through the NLB proxy endpoints from the routes cache).
+    # Only needed for clusters using a non-standard native transport port.
+    #
+    # Required: no
+    # Modifiable at runtime: no
+    # Overridable in a profile: no
+    native-transport-port = 9042
+
   }
 
   # Whether to resolve the addresses passed to `basic.contact-points`.

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

@@ -1161,38 +1161,18 @@ public void should_resolve_to_fallback_when_no_route_for_host_id() {
   // ---- savePort() --------------------------------------------------------
 
   @Test
-  public void savePort_should_use_route_port_when_routes_available() {
-    UUID id1 = UUID.randomUUID();
-    UUID id2 = UUID.randomUUID();
-    handler.setRoutes(
-        ImmutableMap.of(
-            id1, new ClientRouteRecord(id1, "127.0.0.1", 19042),
-            id2, new ClientRouteRecord(id2, "127.0.0.2", 19042)));
-
-    DriverChannel channel = Mockito.mock(DriverChannel.class);
-    handler.savePort(channel);
-
-    assertThat(handler.port).isEqualTo(19042);
-  }
-
-  @Test
-  public void savePort_should_fall_through_to_super_when_routes_empty() {
-    // routes cache is empty by default
+  public void savePort_should_use_native_transport_port_from_config() {
     DriverChannel channel = Mockito.mock(DriverChannel.class);
-    EndPoint ep = Mockito.mock(EndPoint.class);
-    when(ep.resolve()).thenReturn(new InetSocketAddress("127.0.0.1", 9042));
-    when(channel.getEndPoint()).thenReturn(ep);
-
     handler.savePort(channel);
 
+    // Should use the default native transport port from ClientRoutesConfig (9042),
+    // NOT the NLB proxy port from the routes cache or the channel endpoint.
     assertThat(handler.port).isEqualTo(9042);
   }
 
   @Test
   public void savePort_should_skip_when_port_already_set() {
     handler.port = 12345;
-    UUID id = UUID.randomUUID();
-    handler.setRoutes(ImmutableMap.of(id, new ClientRouteRecord(id, "127.0.0.1", 19042)));
 
     DriverChannel channel = Mockito.mock(DriverChannel.class);
     handler.savePort(channel);