Skip to content

Commit a07ad2d

Browse files
nikagranikagra
committed
feat: Client Routes Phase 2 - Core infrastructure 🏗️
- Protocol layer: CLIENT_ROUTES_CHANGE event registration - API: AddressTranslator V2 with Host ID, datacenter, and rack parameters - Infrastructure: ClientRoutesHandler, ClientRoutesAddressTranslator (stubs) - Integration: DefaultDriverContext, DefaultSession, DefaultTopologyMonitor - Tests: ClientRoutesConfigTest, ClientRoutesSessionBuilderTest Phase 2 establishes API. Phase 3 implements queries and lifecycle. Related: DRIVER-88
1 parent fbc2eaf commit a07ad2d

11 files changed

Lines changed: 601 additions & 9 deletions

File tree

‎core/src/main/java/com/datastax/oss/driver/api/core/addresstranslation/AddressTranslator.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@
1818
package com.datastax.oss.driver.api.core.addresstranslation;
1919

2020
import edu.umd.cs.findbugs.annotations.NonNull;
21+
import edu.umd.cs.findbugs.annotations.Nullable;
2122
import java.net.InetSocketAddress;
23+
import java.util.UUID;
2224

2325
/**
2426
* Translates IP addresses received from Cassandra nodes into locally queriable addresses.
@@ -46,6 +48,32 @@ public interface AddressTranslator extends AutoCloseable {
4648
@NonNull
4749
InetSocketAddress translate(@NonNull InetSocketAddress address);
4850

51+
/**
52+
* Translates an address reported by a Cassandra node into the address that the driver will use to
53+
* connect, with additional node metadata for context.
54+
*
55+
* <p>This method is called during node discovery and allows implementations to use the node's
56+
* host ID, datacenter, and rack to make translation decisions. For example, the client routes
57+
* handler uses the host ID to look up the appropriate endpoint mapping.
58+
*
59+
* <p>The default implementation delegates to {@link #translate(InetSocketAddress)}, ignoring the
60+
* additional parameters. This ensures backward compatibility with existing implementations.
61+
*
62+
* @param address the broadcast RPC address of the node
63+
* @param hostId the unique identifier of the node (may be null for contact points)
64+
* @param datacenter the datacenter of the node (may be null if not yet known)
65+
* @param rack the rack of the node (may be null if not yet known)
66+
* @return the translated address
67+
*/
68+
@NonNull
69+
default InetSocketAddress translate(
70+
@NonNull InetSocketAddress address,
71+
@Nullable UUID hostId,
72+
@Nullable String datacenter,
73+
@Nullable String rack) {
74+
return translate(address);
75+
}
76+
4977
/** Called when the cluster that this translator is associated with closes. */
5078
@Override
5179
void close();

‎core/src/main/java/com/datastax/oss/driver/internal/core/clientroutes/ClientRouteInfo.java‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
/*
20+
* Copyright (C) 2025 ScyllaDB
21+
*
22+
* Modified by ScyllaDB
23+
*/
24+
package com.datastax.oss.driver.internal.core.clientroutes;
25+
26+
import edu.umd.cs.findbugs.annotations.NonNull;
27+
import edu.umd.cs.findbugs.annotations.Nullable;
28+
import java.util.Objects;
29+
import java.util.UUID;
30+
import net.jcip.annotations.Immutable;
31+
32+
/**
33+
* Represents a single row from the system.client_routes table.
34+
*
35+
* <p>Each row maps a connection_id + host_id pair to an address that must be DNS-resolved before
36+
* use.
37+
*/
38+
@Immutable
39+
public class ClientRouteInfo {
40+
41+
private final UUID connectionId;
42+
private final UUID hostId;
43+
private final String address;
44+
private final int nativeTransportPort;
45+
private final Integer nativeTransportPortSsl;
46+
47+
public ClientRouteInfo(
48+
@NonNull UUID connectionId,
49+
@NonNull UUID hostId,
50+
@NonNull String address,
51+
int nativeTransportPort,
52+
@Nullable Integer nativeTransportPortSsl) {
53+
this.connectionId = Objects.requireNonNull(connectionId, "connectionId must not be null");
54+
this.hostId = Objects.requireNonNull(hostId, "hostId must not be null");
55+
this.address = Objects.requireNonNull(address, "address must not be null");
56+
this.nativeTransportPort = nativeTransportPort;
57+
this.nativeTransportPortSsl = nativeTransportPortSsl;
58+
}
59+
60+
@NonNull
61+
public UUID getConnectionId() {
62+
return connectionId;
63+
}
64+
65+
@NonNull
66+
public UUID getHostId() {
67+
return hostId;
68+
}
69+
70+
@NonNull
71+
public String getAddress() {
72+
return address;
73+
}
74+
75+
public int getNativeTransportPort() {
76+
return nativeTransportPort;
77+
}
78+
79+
@Nullable
80+
public Integer getNativeTransportPortSsl() {
81+
return nativeTransportPortSsl;
82+
}
83+
84+
@Override
85+
public boolean equals(Object o) {
86+
if (this == o) {
87+
return true;
88+
}
89+
if (!(o instanceof ClientRouteInfo)) {
90+
return false;
91+
}
92+
ClientRouteInfo that = (ClientRouteInfo) o;
93+
return nativeTransportPort == that.nativeTransportPort
94+
&& connectionId.equals(that.connectionId)
95+
&& hostId.equals(that.hostId)
96+
&& address.equals(that.address)
97+
&& Objects.equals(nativeTransportPortSsl, that.nativeTransportPortSsl);
98+
}
99+
100+
@Override
101+
public int hashCode() {
102+
return Objects.hash(connectionId, hostId, address, nativeTransportPort, nativeTransportPortSsl);
103+
}
104+
105+
@Override
106+
public String toString() {
107+
return "ClientRouteInfo{"
108+
+ "connectionId="
109+
+ connectionId
110+
+ ", hostId="
111+
+ hostId
112+
+ ", address='"
113+
+ address
114+
+ '\''
115+
+ ", nativeTransportPort="
116+
+ nativeTransportPort
117+
+ ", nativeTransportPortSsl="
118+
+ nativeTransportPortSsl
119+
+ '}';
120+
}
121+
}

‎core/src/main/java/com/datastax/oss/driver/internal/core/clientroutes/ClientRoutesAddressTranslator.java‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package com.datastax.oss.driver.internal.core.clientroutes;
19+
20+
import com.datastax.oss.driver.api.core.addresstranslation.AddressTranslator;
21+
import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
22+
import edu.umd.cs.findbugs.annotations.NonNull;
23+
import edu.umd.cs.findbugs.annotations.Nullable;
24+
import java.net.InetSocketAddress;
25+
import java.util.UUID;
26+
import net.jcip.annotations.ThreadSafe;
27+
import org.slf4j.Logger;
28+
import org.slf4j.LoggerFactory;
29+
30+
@ThreadSafe
31+
public class ClientRoutesAddressTranslator implements AddressTranslator {
32+
private static final Logger LOG = LoggerFactory.getLogger(ClientRoutesAddressTranslator.class);
33+
private final InternalDriverContext context;
34+
private final String logPrefix;
35+
36+
public ClientRoutesAddressTranslator(@NonNull InternalDriverContext context) {
37+
this.context = context;
38+
this.logPrefix = context.getSessionName();
39+
}
40+
41+
@NonNull
42+
@Override
43+
public InetSocketAddress translate(@NonNull InetSocketAddress address) {
44+
// Contact points bypass translation
45+
return address;
46+
}
47+
48+
@NonNull
49+
@Override
50+
public InetSocketAddress translate(
51+
@NonNull InetSocketAddress address,
52+
@Nullable UUID hostId,
53+
@Nullable String datacenter,
54+
@Nullable String rack) {
55+
// Contact points (hostId == null) bypass translation
56+
if (hostId == null) {
57+
return address;
58+
}
59+
60+
// Lazily get the handler from context to avoid initialization order issues
61+
ClientRoutesHandler handler = context.getClientRoutesHandler();
62+
if (handler == null) {
63+
// This shouldn't happen if ClientRoutesAddressTranslator is only used when configured,
64+
// but handle gracefully just in case
65+
LOG.warn("[{}] ClientRoutesHandler is null, using original address", logPrefix);
66+
return address;
67+
}
68+
69+
boolean useSsl = context.getSslEngineFactory().isPresent();
70+
InetSocketAddress translated = handler.translate(hostId, useSsl);
71+
if (translated != null) {
72+
LOG.debug(
73+
"[{}] Translated {}:{} (host_id={}) to {}:{}",
74+
logPrefix,
75+
address.getAddress().getHostAddress(),
76+
address.getPort(),
77+
hostId,
78+
translated.getAddress().getHostAddress(),
79+
translated.getPort());
80+
return translated;
81+
}
82+
// Fallback to original address if no route found
83+
LOG.debug(
84+
"[{}] No client route found for host_id={}, using original address {}:{}",
85+
logPrefix,
86+
hostId,
87+
address.getAddress().getHostAddress(),
88+
address.getPort());
89+
return address;
90+
}
91+
92+
@Override
93+
public void close() {
94+
// Handler lifecycle is managed by DefaultDriverContext, not by this translator.
95+
// Multiple components may reference the handler, so we don't close it here.
96+
// The handler will be closed when the session closes through the context's cleanup.
97+
}
98+
}

‎core/src/main/java/com/datastax/oss/driver/internal/core/clientroutes/ClientRoutesHandler.java‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package com.datastax.oss.driver.internal.core.clientroutes;
19+
20+
import com.datastax.oss.driver.api.core.config.ClientRoutesConfig;
21+
import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
22+
import edu.umd.cs.findbugs.annotations.NonNull;
23+
import edu.umd.cs.findbugs.annotations.Nullable;
24+
import java.net.InetAddress;
25+
import java.net.InetSocketAddress;
26+
import java.net.UnknownHostException;
27+
import java.util.Map;
28+
import java.util.UUID;
29+
import java.util.concurrent.CompletableFuture;
30+
import java.util.concurrent.CompletionStage;
31+
import java.util.concurrent.ConcurrentHashMap;
32+
import java.util.concurrent.atomic.AtomicReference;
33+
import net.jcip.annotations.ThreadSafe;
34+
import org.slf4j.Logger;
35+
import org.slf4j.LoggerFactory;
36+
37+
@ThreadSafe
38+
public class ClientRoutesHandler implements AutoCloseable {
39+
private static final Logger LOG = LoggerFactory.getLogger(ClientRoutesHandler.class);
40+
41+
@SuppressWarnings("UnusedVariable") // Will be used for querying system.client_routes
42+
private final InternalDriverContext context;
43+
44+
private final ClientRoutesConfig config;
45+
private final String logPrefix;
46+
private final AtomicReference<Map<UUID, ResolvedClientRoute>> resolvedRoutesRef;
47+
private final Map<String, DnsCacheEntry> dnsCache;
48+
private final long dnsCacheDurationNanos;
49+
private volatile boolean closed = false;
50+
51+
public ClientRoutesHandler(
52+
@NonNull InternalDriverContext context, @NonNull ClientRoutesConfig config) {
53+
this.context = context;
54+
this.config = config;
55+
this.logPrefix = context.getSessionName();
56+
this.resolvedRoutesRef = new AtomicReference<>(new ConcurrentHashMap<>());
57+
this.dnsCache = new ConcurrentHashMap<>();
58+
this.dnsCacheDurationNanos = config.getDnsCacheDurationMillis() * 1_000_000L;
59+
}
60+
61+
public CompletionStage<Void> init() {
62+
LOG.debug(
63+
"[{}] Initializing ClientRoutesHandler with {} endpoints",
64+
logPrefix,
65+
config.getEndpoints().size());
66+
return queryAndResolveRoutes();
67+
}
68+
69+
public CompletionStage<Void> refresh() {
70+
LOG.debug("[{}] Refreshing client routes", logPrefix);
71+
return queryAndResolveRoutes();
72+
}
73+
74+
private CompletionStage<Void> queryAndResolveRoutes() {
75+
// TODO: Query system.client_routes table
76+
// For now, return completed future
77+
return CompletableFuture.completedFuture(null);
78+
}
79+
80+
@Nullable
81+
public InetSocketAddress translate(@NonNull UUID hostId, boolean useSsl) {
82+
if (closed) {
83+
return null;
84+
}
85+
Map<UUID, ResolvedClientRoute> routes = resolvedRoutesRef.get();
86+
ResolvedClientRoute route = routes.get(hostId);
87+
if (route == null) {
88+
LOG.debug("[{}] No client route found for host_id={}", logPrefix, hostId);
89+
return null;
90+
}
91+
return route.toSocketAddress(useSsl);
92+
}
93+
94+
@SuppressWarnings("UnusedMethod") // Will be used when implementing system.client_routes query
95+
private InetAddress resolveDns(String hostname) throws UnknownHostException {
96+
DnsCacheEntry cached = dnsCache.get(hostname);
97+
long now = System.nanoTime();
98+
if (cached != null && (now - cached.resolvedAtNanos) < dnsCacheDurationNanos) {
99+
return cached.address;
100+
}
101+
InetAddress resolved = InetAddress.getByName(hostname);
102+
dnsCache.put(hostname, new DnsCacheEntry(resolved, now));
103+
return resolved;
104+
}
105+
106+
@Override
107+
public void close() {
108+
closed = true;
109+
dnsCache.clear();
110+
LOG.debug("[{}] ClientRoutesHandler closed", logPrefix);
111+
}
112+
113+
private static class DnsCacheEntry {
114+
final InetAddress address;
115+
final long resolvedAtNanos;
116+
117+
DnsCacheEntry(InetAddress address, long resolvedAtNanos) {
118+
this.address = address;
119+
this.resolvedAtNanos = resolvedAtNanos;
120+
}
121+
}
122+
}

0 commit comments

Comments
 (0)