com.datastax.oss.driver.api.core.context.DriverContext::getRequestIdGenerator()",
+ "justification": "CASSJAVA-97: Let users inject an ID for each request and write to the custom payload"
}
]
}
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java
index c5a482c5d50..56a6fe9c2be 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java
@@ -1119,7 +1119,61 @@ public enum DefaultDriverOption implements DriverOption {
*
* 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 or not to do a DNS reverse-lookup of provided server addresses for SAN addresses.
+ *
+ *
Value-type: boolean
+ */
+ SSL_ALLOW_DNS_REVERSE_LOOKUP_SAN("advanced.ssl-engine-factory.allow-dns-reverse-lookup-san"),
+ /**
+ * The class of session-wide component that generates request IDs.
+ *
+ *
Value-type: {@link String}
+ */
+ REQUEST_ID_GENERATOR_CLASS("advanced.request-id.generator.class"),
+ /**
+ * An address to always translate all node addresses to that same proxy hostname no matter what IP
+ * address a node has, but still using its native transport port.
+ *
+ *
Value-Type: {@link String}
+ */
+ ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME("advanced.address-translator.advertised-hostname"),
+ /**
+ * A map of Cassandra node subnets (CIDR notations) to target addresses, for example (note quoted
+ * keys):
+ *
+ *
+ * advanced.address-translator.subnet-addresses {
+ * "100.64.0.0/15" = "cassandra.datacenter1.com:9042"
+ * "100.66.0.0/15" = "cassandra.datacenter2.com:9042"
+ * # IPv6 example:
+ * # "::ffff:6440:0/111" = "cassandra.datacenter1.com:9042"
+ * # "::ffff:6442:0/111" = "cassandra.datacenter2.com:9042"
+ * }
+ *
+ *
+ * Note: subnets must be represented as prefix blocks, see {@code
+ * inet.ipaddr.Address#isPrefixBlock()}.
+ *
+ * Value type: {@link java.util.Map Map}<{@link String},{@link String}>
+ */
+ ADDRESS_TRANSLATOR_SUBNET_ADDRESSES("advanced.address-translator.subnet-addresses"),
+ /**
+ * A default address to fallback to if Cassandra node IP isn't contained in any of the configured
+ * subnets.
+ *
+ *
Value-Type: {@link String}
+ */
+ ADDRESS_TRANSLATOR_DEFAULT_ADDRESS("advanced.address-translator.default-address"),
+ /**
+ * Whether to resolve the addresses on initialization (if true) or on each node (re-)connection
+ * (if false). Defaults to false.
+ *
+ *
Value-Type: boolean
+ */
+ ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses");
private final String path;
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java
index db5edb5b947..e7c606cade8 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java
@@ -248,6 +248,10 @@ public String toString() {
*/
public static final TypedDriverOption SSL_HOSTNAME_VALIDATION =
new TypedDriverOption<>(DefaultDriverOption.SSL_HOSTNAME_VALIDATION, GenericType.BOOLEAN);
+
+ public static final TypedDriverOption SSL_ALLOW_DNS_REVERSE_LOOKUP_SAN =
+ new TypedDriverOption<>(
+ DefaultDriverOption.SSL_ALLOW_DNS_REVERSE_LOOKUP_SAN, GenericType.BOOLEAN);
/** The location of the keystore file. */
public static final TypedDriverOption SSL_KEYSTORE_PATH =
new TypedDriverOption<>(DefaultDriverOption.SSL_KEYSTORE_PATH, GenericType.STRING);
@@ -296,6 +300,10 @@ public String toString() {
new TypedDriverOption<>(
DefaultDriverOption.REQUEST_TRACKER_CLASSES, GenericType.listOf(String.class));
+ /** The class of a session-wide component that generates request IDs. */
+ public static final TypedDriverOption REQUEST_ID_GENERATOR_CLASS =
+ new TypedDriverOption<>(DefaultDriverOption.REQUEST_ID_GENERATOR_CLASS, GenericType.STRING);
+
/** Whether to log successful requests. */
public static final TypedDriverOption REQUEST_LOGGER_SUCCESS_ENABLED =
new TypedDriverOption<>(
@@ -923,6 +931,20 @@ public String toString() {
DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_ALLOW_FOR_LOCAL_CONSISTENCY_LEVELS,
GenericType.BOOLEAN);
+ public static final TypedDriverOption ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME =
+ new TypedDriverOption<>(
+ DefaultDriverOption.ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME, GenericType.STRING);
+ public static final TypedDriverOption> ADDRESS_TRANSLATOR_SUBNET_ADDRESSES =
+ new TypedDriverOption<>(
+ DefaultDriverOption.ADDRESS_TRANSLATOR_SUBNET_ADDRESSES,
+ GenericType.mapOf(GenericType.STRING, GenericType.STRING));
+ public static final TypedDriverOption ADDRESS_TRANSLATOR_DEFAULT_ADDRESS =
+ new TypedDriverOption<>(
+ DefaultDriverOption.ADDRESS_TRANSLATOR_DEFAULT_ADDRESS, GenericType.STRING);
+ public static final TypedDriverOption ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES =
+ new TypedDriverOption<>(
+ DefaultDriverOption.ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES, GenericType.BOOLEAN);
+
/**
* Ordered preference list of remote dcs optionally supplied for automatic failover and included
* in query plan. This feature is enabled only when max-nodes-per-remote-dc is greater than 0.
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/context/DriverContext.java b/core/src/main/java/com/datastax/oss/driver/api/core/context/DriverContext.java
index 5b32389e362..6f0afd3df8a 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/context/DriverContext.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/context/DriverContext.java
@@ -33,6 +33,7 @@
import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy;
import com.datastax.oss.driver.api.core.ssl.SslEngineFactory;
import com.datastax.oss.driver.api.core.time.TimestampGenerator;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
import com.datastax.oss.driver.api.core.tracker.RequestTracker;
import edu.umd.cs.findbugs.annotations.NonNull;
import java.util.Map;
@@ -139,6 +140,10 @@ default SpeculativeExecutionPolicy getSpeculativeExecutionPolicy(@NonNull String
@NonNull
RequestTracker getRequestTracker();
+ /** @return The driver's request ID generator; never {@code null}. */
+ @NonNull
+ Optional getRequestIdGenerator();
+
/** @return The driver's request throttler; never {@code null}. */
@NonNull
RequestThrottler getRequestThrottler();
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LoadBalancingPolicy.java b/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LoadBalancingPolicy.java
index d890ae6c100..de0d9db4ebd 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LoadBalancingPolicy.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LoadBalancingPolicy.java
@@ -24,6 +24,7 @@
import com.datastax.oss.driver.api.core.tracker.RequestTracker;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.Nullable;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.Queue;
@@ -76,6 +77,12 @@ default Optional getRequestTracker() {
*/
void init(@NonNull Map nodes, @NonNull DistanceReporter distanceReporter);
+ /** Returns map containing details that impact C* node connectivity. */
+ @NonNull
+ default Map getStartupConfiguration() {
+ return Collections.emptyMap();
+ }
+
/**
* Returns the coordinators to use for a new query.
*
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/session/ProgrammaticArguments.java b/core/src/main/java/com/datastax/oss/driver/api/core/session/ProgrammaticArguments.java
index 0373831da41..4db44655dc2 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/session/ProgrammaticArguments.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/session/ProgrammaticArguments.java
@@ -24,6 +24,7 @@
import com.datastax.oss.driver.api.core.metadata.NodeStateListener;
import com.datastax.oss.driver.api.core.metadata.schema.SchemaChangeListener;
import com.datastax.oss.driver.api.core.ssl.SslEngineFactory;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
import com.datastax.oss.driver.api.core.tracker.RequestTracker;
import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
import com.datastax.oss.driver.api.core.type.codec.registry.MutableCodecRegistry;
@@ -60,6 +61,7 @@ public static Builder builder() {
private final NodeStateListener nodeStateListener;
private final SchemaChangeListener schemaChangeListener;
private final RequestTracker requestTracker;
+ private final RequestIdGenerator requestIdGenerator;
private final Map localDatacenters;
private final Map> nodeFilters;
private final Map nodeDistanceEvaluators;
@@ -79,6 +81,7 @@ private ProgrammaticArguments(
@Nullable NodeStateListener nodeStateListener,
@Nullable SchemaChangeListener schemaChangeListener,
@Nullable RequestTracker requestTracker,
+ @Nullable RequestIdGenerator requestIdGenerator,
@NonNull Map localDatacenters,
@NonNull Map> nodeFilters,
@NonNull Map nodeDistanceEvaluators,
@@ -97,6 +100,7 @@ private ProgrammaticArguments(
this.nodeStateListener = nodeStateListener;
this.schemaChangeListener = schemaChangeListener;
this.requestTracker = requestTracker;
+ this.requestIdGenerator = requestIdGenerator;
this.localDatacenters = localDatacenters;
this.nodeFilters = nodeFilters;
this.nodeDistanceEvaluators = nodeDistanceEvaluators;
@@ -132,6 +136,11 @@ public RequestTracker getRequestTracker() {
return requestTracker;
}
+ @Nullable
+ public RequestIdGenerator getRequestIdGenerator() {
+ return requestIdGenerator;
+ }
+
@NonNull
public Map getLocalDatacenters() {
return localDatacenters;
@@ -205,6 +214,7 @@ public static class Builder {
private NodeStateListener nodeStateListener;
private SchemaChangeListener schemaChangeListener;
private RequestTracker requestTracker;
+ private RequestIdGenerator requestIdGenerator;
private ImmutableMap.Builder localDatacentersBuilder = ImmutableMap.builder();
private final ImmutableMap.Builder> nodeFiltersBuilder =
ImmutableMap.builder();
@@ -304,6 +314,12 @@ public Builder addRequestTracker(@NonNull RequestTracker requestTracker) {
return this;
}
+ @NonNull
+ public Builder withRequestIdGenerator(@Nullable RequestIdGenerator requestIdGenerator) {
+ this.requestIdGenerator = requestIdGenerator;
+ return this;
+ }
+
@NonNull
public Builder withLocalDatacenter(
@NonNull String profileName, @NonNull String localDatacenter) {
@@ -433,6 +449,7 @@ public ProgrammaticArguments build() {
nodeStateListener,
schemaChangeListener,
requestTracker,
+ requestIdGenerator,
localDatacentersBuilder.build(),
nodeFiltersBuilder.build(),
nodeDistanceEvaluatorsBuilder.build(),
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java b/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java
index a2c0f933efc..8375f0ef30b 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java
@@ -42,6 +42,7 @@
import com.datastax.oss.driver.api.core.metadata.schema.SchemaChangeListener;
import com.datastax.oss.driver.api.core.ssl.ProgrammaticSslEngineFactory;
import com.datastax.oss.driver.api.core.ssl.SslEngineFactory;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
import com.datastax.oss.driver.api.core.tracker.RequestTracker;
import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
import com.datastax.oss.driver.api.core.type.codec.registry.MutableCodecRegistry;
@@ -54,6 +55,7 @@
import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
import com.datastax.oss.driver.internal.core.session.DefaultSession;
+import com.datastax.oss.driver.internal.core.tracker.W3CContextRequestIdGenerator;
import com.datastax.oss.driver.internal.core.util.concurrent.BlockingOperation;
import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
import edu.umd.cs.findbugs.annotations.NonNull;
@@ -90,6 +92,8 @@
@NotThreadSafe
public abstract class SessionBuilder {
+ public static final String ASTRA_PAYLOAD_KEY = "traceparent";
+
private static final Logger LOG = LoggerFactory.getLogger(SessionBuilder.class);
@SuppressWarnings("unchecked")
@@ -324,6 +328,17 @@ public SelfT addRequestTracker(@NonNull RequestTracker requestTracker) {
return self;
}
+ /**
+ * Registers a request ID generator. The driver will use the generated ID in the logs and
+ * optionally add to the custom payload so that users can correlate logs about the same request
+ * from the Cassandra side.
+ */
+ @NonNull
+ public SelfT withRequestIdGenerator(@NonNull RequestIdGenerator requestIdGenerator) {
+ this.programmaticArgumentsBuilder.withRequestIdGenerator(requestIdGenerator);
+ return self;
+ }
+
/**
* Registers an authentication provider to use with the session.
*
@@ -905,6 +920,13 @@ protected final CompletionStage buildDefaultSessionAsync() {
List configContactPoints =
defaultConfig.getStringList(DefaultDriverOption.CONTACT_POINTS, Collections.emptyList());
if (cloudConfigInputStream != null) {
+ // override request id generator, unless user has already set it
+ if (programmaticArguments.getRequestIdGenerator() == null) {
+ programmaticArgumentsBuilder.withRequestIdGenerator(
+ new W3CContextRequestIdGenerator(ASTRA_PAYLOAD_KEY));
+ LOG.debug(
+ "A secure connect bundle is provided, using W3CContextRequestIdGenerator as request ID generator.");
+ }
if (!programmaticContactPoints.isEmpty() || !configContactPoints.isEmpty()) {
LOG.info(
"Both a secure connect bundle and contact points were provided. These are mutually exclusive. The contact points from the secure bundle will have priority.");
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/session/throttling/RequestThrottler.java b/core/src/main/java/com/datastax/oss/driver/api/core/session/throttling/RequestThrottler.java
index 7e2b41ebbdb..73d347d533e 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/session/throttling/RequestThrottler.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/session/throttling/RequestThrottler.java
@@ -23,10 +23,10 @@
/**
* Limits the number of concurrent requests executed by the driver.
*
- * Usage in non-blocking applications: beware that all built-in implementations of this interface
- * use locks for internal coordination, and do not qualify as lock-free, with the obvious exception
- * of {@code PassThroughRequestThrottler}. If your application enforces strict lock-freedom, then
- * request throttling should not be enabled.
+ *
Usage in non-blocking applications: beware that some implementations of this interface use
+ * locks for internal coordination, and do not qualify as lock-free. If your application enforces
+ * strict lock-freedom, then you should use the {@code PassThroughRequestThrottler} or the {@code
+ * ConcurrencyLimitingRequestThrottler}.
*/
public interface RequestThrottler extends Closeable {
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/ssl/ProgrammaticSslEngineFactory.java b/core/src/main/java/com/datastax/oss/driver/api/core/ssl/ProgrammaticSslEngineFactory.java
index 6dfe4087b91..d65eaa864aa 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/ssl/ProgrammaticSslEngineFactory.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/ssl/ProgrammaticSslEngineFactory.java
@@ -45,6 +45,7 @@ public class ProgrammaticSslEngineFactory implements SslEngineFactory {
protected final SSLContext sslContext;
protected final String[] cipherSuites;
protected final boolean requireHostnameValidation;
+ protected final boolean allowDnsReverseLookupSan;
/**
* Creates an instance with the given {@link SSLContext}, default cipher suites and no host name
@@ -80,9 +81,28 @@ public ProgrammaticSslEngineFactory(
@NonNull SSLContext sslContext,
@Nullable String[] cipherSuites,
boolean requireHostnameValidation) {
+ this(sslContext, cipherSuites, requireHostnameValidation, true);
+ }
+
+ /**
+ * Creates an instance with the given {@link SSLContext}, cipher suites and host name validation.
+ *
+ * @param sslContext the {@link SSLContext} to use.
+ * @param cipherSuites the cipher suites to use, or null to use the default ones.
+ * @param requireHostnameValidation whether to enable host name validation. If enabled, host name
+ * validation will be done using HTTPS algorithm.
+ * @param allowDnsReverseLookupSan whether to allow raw server IPs to be DNS reverse-resolved to
+ * choose the appropriate Subject Alternative Name.
+ */
+ public ProgrammaticSslEngineFactory(
+ @NonNull SSLContext sslContext,
+ @Nullable String[] cipherSuites,
+ boolean requireHostnameValidation,
+ boolean allowDnsReverseLookupSan) {
this.sslContext = sslContext;
this.cipherSuites = cipherSuites;
this.requireHostnameValidation = requireHostnameValidation;
+ this.allowDnsReverseLookupSan = allowDnsReverseLookupSan;
}
@NonNull
@@ -92,7 +112,12 @@ public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) {
SocketAddress remoteAddress = remoteEndpoint.resolve();
if (remoteAddress instanceof InetSocketAddress) {
InetSocketAddress socketAddress = (InetSocketAddress) remoteAddress;
- engine = sslContext.createSSLEngine(socketAddress.getHostName(), socketAddress.getPort());
+ engine =
+ sslContext.createSSLEngine(
+ allowDnsReverseLookupSan
+ ? socketAddress.getHostName()
+ : socketAddress.getHostString(),
+ socketAddress.getPort());
} else {
engine = sslContext.createSSLEngine();
}
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/tracker/RequestIdGenerator.java b/core/src/main/java/com/datastax/oss/driver/api/core/tracker/RequestIdGenerator.java
new file mode 100644
index 00000000000..21db3793b01
--- /dev/null
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/tracker/RequestIdGenerator.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.api.core.tracker;
+
+import com.datastax.oss.driver.api.core.cql.Statement;
+import com.datastax.oss.driver.api.core.session.Request;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Interface responsible for generating request IDs.
+ *
+ *
Note that all request IDs have a parent/child relationship. A "session request ID" can loosely
+ * be thought of as encompassing a sequence of a request + any attendant retries, speculative
+ * executions etc. It's scope is identical to that of a {@link
+ * com.datastax.oss.driver.internal.core.cql.CqlRequestHandler}. A "node request ID" represents a
+ * single request within this larger scope. Note that a request corresponding to a request ID may be
+ * retried; in that case the retry count will be appended to the corresponding identifier in the
+ * logs.
+ */
+public interface RequestIdGenerator {
+
+ String DEFAULT_PAYLOAD_KEY = "request-id";
+
+ /**
+ * Generates a unique identifier for the session request. This will be the identifier for the
+ * entire `session.execute()` call. This identifier will be added to logs, and propagated to
+ * request trackers.
+ *
+ * @return a unique identifier for the session request
+ */
+ String getSessionRequestId();
+
+ /**
+ * Generates a unique identifier for the node request. This will be the identifier for the CQL
+ * request against a particular node. There can be one or more node requests for a single session
+ * request, due to retries or speculative executions. This identifier will be added to logs, and
+ * propagated to request trackers.
+ *
+ * @param statement the statement to be executed
+ * @param parentId the session request identifier
+ * @return a unique identifier for the node request
+ */
+ String getNodeRequestId(@NonNull Request statement, @NonNull String parentId);
+
+ default String getCustomPayloadKey() {
+ return DEFAULT_PAYLOAD_KEY;
+ }
+
+ default Statement getDecoratedStatement(
+ @NonNull Statement statement, @NonNull String requestId) {
+
+ Map existing = new HashMap<>(statement.getCustomPayload());
+ String key = getCustomPayloadKey();
+
+ // Add or overwrite
+ existing.put(key, ByteBuffer.wrap(requestId.getBytes(StandardCharsets.UTF_8)));
+
+ // Allowing null key/values
+ // Wrap a map inside to be immutable without instanciating a new map
+ Map unmodifiableMap = Collections.unmodifiableMap(existing);
+
+ return statement.setCustomPayload(unmodifiableMap);
+ }
+}
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/tracker/RequestTracker.java b/core/src/main/java/com/datastax/oss/driver/api/core/tracker/RequestTracker.java
index d29ee48d352..065b41e496a 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/tracker/RequestTracker.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/tracker/RequestTracker.java
@@ -47,21 +47,22 @@ default void onSuccess(
@NonNull Node node) {}
/**
- * Invoked each time a request succeeds.
+ * Invoked each time a session request succeeds. A session request is a `session.execute()` call
*
* @param latencyNanos the overall execution time (from the {@link Session#execute(Request,
* GenericType) session.execute} call until the result is made available to the client).
* @param executionProfile the execution profile of this request.
* @param node the node that returned the successful response.
- * @param requestLogPrefix the dedicated log prefix for this request
+ * @param sessionRequestLogPrefix the dedicated log prefix for this request
*/
default void onSuccess(
@NonNull Request request,
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String requestLogPrefix) {
- // If client doesn't override onSuccess with requestLogPrefix delegate call to the old method
+ @NonNull String sessionRequestLogPrefix) {
+ // If client doesn't override onSuccess with sessionRequestLogPrefix delegate call to the old
+ // method
onSuccess(request, latencyNanos, executionProfile, node);
}
@@ -78,13 +79,13 @@ default void onError(
@Nullable Node node) {}
/**
- * Invoked each time a request fails.
+ * Invoked each time a session request fails. A session request is a `session.execute()` call
*
* @param latencyNanos the overall execution time (from the {@link Session#execute(Request,
* GenericType) session.execute} call until the error is propagated to the client).
* @param executionProfile the execution profile of this request.
* @param node the node that returned the error response, or {@code null} if the error occurred
- * @param requestLogPrefix the dedicated log prefix for this request
+ * @param sessionRequestLogPrefix the dedicated log prefix for this request
*/
default void onError(
@NonNull Request request,
@@ -92,8 +93,9 @@ default void onError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@Nullable Node node,
- @NonNull String requestLogPrefix) {
- // If client doesn't override onError with requestLogPrefix delegate call to the old method
+ @NonNull String sessionRequestLogPrefix) {
+ // If client doesn't override onError with sessionRequestLogPrefix delegate call to the old
+ // method
onError(request, error, latencyNanos, executionProfile, node);
}
@@ -110,14 +112,15 @@ default void onNodeError(
@NonNull Node node) {}
/**
- * Invoked each time a request fails at the node level. Similar to {@link #onError(Request,
- * Throwable, long, DriverExecutionProfile, Node, String)} but at a per node level.
+ * Invoked each time a node request fails. A node request is a CQL request sent to a particular
+ * node. There can be one or more node requests for a single session request, due to retries or
+ * speculative executions.
*
* @param latencyNanos the overall execution time (from the {@link Session#execute(Request,
* GenericType) session.execute} call until the error is propagated to the client).
* @param executionProfile the execution profile of this request.
* @param node the node that returned the error response.
- * @param requestLogPrefix the dedicated log prefix for this request
+ * @param nodeRequestLogPrefix the dedicated log prefix for this request
*/
default void onNodeError(
@NonNull Request request,
@@ -125,8 +128,9 @@ default void onNodeError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String requestLogPrefix) {
- // If client doesn't override onNodeError with requestLogPrefix delegate call to the old method
+ @NonNull String nodeRequestLogPrefix) {
+ // If client doesn't override onNodeError with nodeRequestLogPrefix delegate call to the old
+ // method
onNodeError(request, error, latencyNanos, executionProfile, node);
}
@@ -142,22 +146,23 @@ default void onNodeSuccess(
@NonNull Node node) {}
/**
- * Invoked each time a request succeeds at the node level. Similar to {@link #onSuccess(Request,
- * long, DriverExecutionProfile, Node, String)} but at per node level.
+ * Invoked each time a node request succeeds. A node request is a CQL request sent to a particular
+ * node. There can be one or more node requests for a single session request, due to retries or
+ * speculative executions.
*
* @param latencyNanos the overall execution time (from the {@link Session#execute(Request,
* GenericType) session.execute} call until the result is made available to the client).
* @param executionProfile the execution profile of this request.
* @param node the node that returned the successful response.
- * @param requestLogPrefix the dedicated log prefix for this request
+ * @param nodeRequestLogPrefix the dedicated log prefix for this request
*/
default void onNodeSuccess(
@NonNull Request request,
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String requestLogPrefix) {
- // If client doesn't override onNodeSuccess with requestLogPrefix delegate call to the old
+ @NonNull String nodeRequestLogPrefix) {
+ // If client doesn't override onNodeSuccess with nodeRequestLogPrefix delegate call to the old
// method
onNodeSuccess(request, latencyNanos, executionProfile, node);
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/ContactPoints.java b/core/src/main/java/com/datastax/oss/driver/internal/core/ContactPoints.java
index 1ed2a1cebf3..bb65661b72f 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/ContactPoints.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/ContactPoints.java
@@ -19,14 +19,11 @@
import com.datastax.oss.driver.api.core.metadata.EndPoint;
import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
+import com.datastax.oss.driver.internal.core.util.AddressUtils;
import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet;
import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
-import java.net.InetAddress;
import java.net.InetSocketAddress;
-import java.net.UnknownHostException;
-import java.util.Arrays;
import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.slf4j.Logger;
@@ -41,7 +38,22 @@ public static Set merge(
Set result = Sets.newHashSet(programmaticContactPoints);
for (String spec : configContactPoints) {
- for (InetSocketAddress address : extract(spec, resolve)) {
+
+ Set addresses = Collections.emptySet();
+ try {
+ addresses = AddressUtils.extract(spec, resolve);
+ } catch (RuntimeException e) {
+ LOG.warn("Ignoring invalid contact point {} ({})", spec, e.getMessage(), e);
+ }
+
+ if (addresses.size() > 1) {
+ LOG.info(
+ "Contact point {} resolves to multiple addresses, will use them all ({})",
+ spec,
+ addresses);
+ }
+
+ for (InetSocketAddress address : addresses) {
DefaultEndPoint endPoint = new DefaultEndPoint(address);
boolean wasNew = result.add(endPoint);
if (!wasNew) {
@@ -51,43 +63,4 @@ public static Set merge(
}
return ImmutableSet.copyOf(result);
}
-
- private static Set extract(String spec, boolean resolve) {
- int separator = spec.lastIndexOf(':');
- if (separator < 0) {
- LOG.warn("Ignoring invalid contact point {} (expecting host:port)", spec);
- return Collections.emptySet();
- }
-
- String host = spec.substring(0, separator);
- String portSpec = spec.substring(separator + 1);
- int port;
- try {
- port = Integer.parseInt(portSpec);
- } catch (NumberFormatException e) {
- LOG.warn("Ignoring invalid contact point {} (expecting a number, got {})", spec, portSpec);
- return Collections.emptySet();
- }
- if (!resolve) {
- return ImmutableSet.of(InetSocketAddress.createUnresolved(host, port));
- } else {
- try {
- InetAddress[] inetAddresses = InetAddress.getAllByName(host);
- if (inetAddresses.length > 1) {
- LOG.info(
- "Contact point {} resolves to multiple addresses, will use them all ({})",
- spec,
- Arrays.deepToString(inetAddresses));
- }
- Set result = new HashSet<>();
- for (InetAddress inetAddress : inetAddresses) {
- result.add(new InetSocketAddress(inetAddress, port));
- }
- return result;
- } catch (UnknownHostException e) {
- LOG.warn("Ignoring invalid contact point {} (unknown host {})", spec, host);
- return Collections.emptySet();
- }
- }
- }
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslator.java b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslator.java
index ea624c3e84d..0ee1e22a7ac 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslator.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslator.java
@@ -17,8 +17,9 @@
*/
package com.datastax.oss.driver.internal.core.addresstranslation;
+import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME;
+
import com.datastax.oss.driver.api.core.addresstranslation.AddressTranslator;
-import com.datastax.oss.driver.api.core.config.DriverOption;
import com.datastax.oss.driver.api.core.context.DriverContext;
import edu.umd.cs.findbugs.annotations.NonNull;
import java.net.InetSocketAddress;
@@ -43,28 +44,13 @@ public class FixedHostNameAddressTranslator implements AddressTranslator {
private static final Logger LOG = LoggerFactory.getLogger(FixedHostNameAddressTranslator.class);
- public static final String ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME =
- "advanced.address-translator.advertised-hostname";
-
- public static DriverOption ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME_OPTION =
- new DriverOption() {
- @NonNull
- @Override
- public String getPath() {
- return ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME;
- }
- };
-
private final String advertisedHostname;
private final String logPrefix;
public FixedHostNameAddressTranslator(@NonNull DriverContext context) {
logPrefix = context.getSessionName();
advertisedHostname =
- context
- .getConfig()
- .getDefaultProfile()
- .getString(ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME_OPTION);
+ context.getConfig().getDefaultProfile().getString(ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME);
}
@NonNull
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/Subnet.java b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/Subnet.java
new file mode 100644
index 00000000000..7c25e94e2f9
--- /dev/null
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/Subnet.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.addresstranslation;
+
+import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
+import com.datastax.oss.driver.shaded.guava.common.base.Splitter;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.List;
+
+class Subnet {
+ private final byte[] subnet;
+ private final byte[] networkMask;
+ private final byte[] upper;
+ private final byte[] lower;
+
+ private Subnet(byte[] subnet, byte[] networkMask) {
+ this.subnet = subnet;
+ this.networkMask = networkMask;
+
+ byte[] upper = new byte[subnet.length];
+ byte[] lower = new byte[subnet.length];
+ for (int i = 0; i < subnet.length; i++) {
+ upper[i] = (byte) (subnet[i] | ~networkMask[i]);
+ lower[i] = (byte) (subnet[i] & networkMask[i]);
+ }
+ this.upper = upper;
+ this.lower = lower;
+ }
+
+ static Subnet parse(String subnetCIDR) throws UnknownHostException {
+ List parts = Splitter.on("/").splitToList(subnetCIDR);
+ if (parts.size() != 2) {
+ throw new IllegalArgumentException("Invalid subnet: " + subnetCIDR);
+ }
+
+ boolean isIPv6 = parts.get(0).contains(":");
+ byte[] subnet = InetAddress.getByName(parts.get(0)).getAddress();
+ if (isIPv4(subnet) && isIPv6) {
+ subnet = toIPv6(subnet);
+ }
+ int prefixLength = Integer.parseInt(parts.get(1));
+ validatePrefixLength(subnet, prefixLength);
+
+ byte[] networkMask = toNetworkMask(subnet, prefixLength);
+ validateSubnetIsPrefixBlock(subnet, networkMask, subnetCIDR);
+ return new Subnet(subnet, networkMask);
+ }
+
+ private static byte[] toNetworkMask(byte[] subnet, int prefixLength) {
+ int fullBytes = prefixLength / 8;
+ int remainingBits = prefixLength % 8;
+ byte[] mask = new byte[subnet.length];
+ Arrays.fill(mask, 0, fullBytes, (byte) 0xFF);
+ if (remainingBits > 0) {
+ mask[fullBytes] = (byte) (0xFF << (8 - remainingBits));
+ }
+ return mask;
+ }
+
+ private static void validatePrefixLength(byte[] subnet, int prefixLength) {
+ int max_prefix_length = subnet.length * 8;
+ if (prefixLength < 0 || max_prefix_length < prefixLength) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Prefix length %s must be within [0; %s]", prefixLength, max_prefix_length));
+ }
+ }
+
+ private static void validateSubnetIsPrefixBlock(
+ byte[] subnet, byte[] networkMask, String subnetCIDR) {
+ byte[] prefixBlock = toPrefixBlock(subnet, networkMask);
+ if (!Arrays.equals(subnet, prefixBlock)) {
+ throw new IllegalArgumentException(
+ String.format("Subnet %s must be represented as a network prefix block", subnetCIDR));
+ }
+ }
+
+ private static byte[] toPrefixBlock(byte[] subnet, byte[] networkMask) {
+ byte[] prefixBlock = new byte[subnet.length];
+ for (int i = 0; i < subnet.length; i++) {
+ prefixBlock[i] = (byte) (subnet[i] & networkMask[i]);
+ }
+ return prefixBlock;
+ }
+
+ @VisibleForTesting
+ byte[] getSubnet() {
+ return Arrays.copyOf(subnet, subnet.length);
+ }
+
+ @VisibleForTesting
+ byte[] getNetworkMask() {
+ return Arrays.copyOf(networkMask, networkMask.length);
+ }
+
+ byte[] getUpper() {
+ return Arrays.copyOf(upper, upper.length);
+ }
+
+ byte[] getLower() {
+ return Arrays.copyOf(lower, lower.length);
+ }
+
+ boolean isIPv4() {
+ return isIPv4(subnet);
+ }
+
+ boolean isIPv6() {
+ return isIPv6(subnet);
+ }
+
+ boolean contains(byte[] ip) {
+ if (isIPv4() && !isIPv4(ip)) {
+ return false;
+ }
+ if (isIPv6() && isIPv4(ip)) {
+ ip = toIPv6(ip);
+ }
+ if (subnet.length != ip.length) {
+ throw new IllegalArgumentException(
+ "IP version is unknown: " + Arrays.toString(toZeroBasedByteArray(ip)));
+ }
+ for (int i = 0; i < subnet.length; i++) {
+ if (subnet[i] != (byte) (ip[i] & networkMask[i])) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean isIPv4(byte[] ip) {
+ return ip.length == 4;
+ }
+
+ private static boolean isIPv6(byte[] ip) {
+ return ip.length == 16;
+ }
+
+ private static byte[] toIPv6(byte[] ipv4) {
+ byte[] ipv6 = new byte[16];
+ ipv6[10] = (byte) 0xFF;
+ ipv6[11] = (byte) 0xFF;
+ System.arraycopy(ipv4, 0, ipv6, 12, 4);
+ return ipv6;
+ }
+
+ @Override
+ public String toString() {
+ return Arrays.toString(toZeroBasedByteArray(subnet));
+ }
+
+ private static int[] toZeroBasedByteArray(byte[] bytes) {
+ int[] res = new int[bytes.length];
+ for (int i = 0; i < bytes.length; i++) {
+ res[i] = bytes[i] & 0xFF;
+ }
+ return res;
+ }
+}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddress.java b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddress.java
new file mode 100644
index 00000000000..105e776a507
--- /dev/null
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddress.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.addresstranslation;
+
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+
+class SubnetAddress {
+ private final Subnet subnet;
+ private final InetSocketAddress address;
+
+ SubnetAddress(String subnetCIDR, InetSocketAddress address) {
+ try {
+ this.subnet = Subnet.parse(subnetCIDR);
+ } catch (UnknownHostException e) {
+ throw new RuntimeException(e);
+ }
+ this.address = address;
+ }
+
+ InetSocketAddress getAddress() {
+ return this.address;
+ }
+
+ boolean isOverlapping(SubnetAddress other) {
+ Subnet thisSubnet = this.subnet;
+ Subnet otherSubnet = other.subnet;
+ return thisSubnet.contains(otherSubnet.getLower())
+ || thisSubnet.contains(otherSubnet.getUpper())
+ || otherSubnet.contains(thisSubnet.getLower())
+ || otherSubnet.contains(thisSubnet.getUpper());
+ }
+
+ boolean contains(InetSocketAddress address) {
+ return subnet.contains(address.getAddress().getAddress());
+ }
+
+ boolean isIPv4() {
+ return subnet.isIPv4();
+ }
+
+ boolean isIPv6() {
+ return subnet.isIPv6();
+ }
+
+ @Override
+ public String toString() {
+ return "SubnetAddress[subnet=" + subnet + ", address=" + address + "]";
+ }
+}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTranslator.java b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTranslator.java
new file mode 100644
index 00000000000..85f29e3fadd
--- /dev/null
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTranslator.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.addresstranslation;
+
+import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.ADDRESS_TRANSLATOR_DEFAULT_ADDRESS;
+import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES;
+import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.ADDRESS_TRANSLATOR_SUBNET_ADDRESSES;
+
+import com.datastax.oss.driver.api.core.addresstranslation.AddressTranslator;
+import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.internal.core.util.AddressUtils;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This translator returns the proxy address of the private subnet containing the Cassandra node IP,
+ * or default address if no matching subnets, or passes through the original node address if no
+ * default configured.
+ *
+ * The translator can be used for scenarios when all nodes are behind some kind of proxy, and
+ * that proxy is different for nodes located in different subnets (eg. when Cassandra is deployed in
+ * multiple datacenters/regions). One can use this, for example, for Cassandra on Kubernetes with
+ * different Cassandra datacenters deployed to different Kubernetes clusters.
+ */
+public class SubnetAddressTranslator implements AddressTranslator {
+ private static final Logger LOG = LoggerFactory.getLogger(SubnetAddressTranslator.class);
+
+ private final List subnetAddresses;
+
+ @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+ private final Optional defaultAddress;
+
+ private final String logPrefix;
+
+ public SubnetAddressTranslator(@NonNull DriverContext context) {
+ logPrefix = context.getSessionName();
+ boolean resolveAddresses =
+ context
+ .getConfig()
+ .getDefaultProfile()
+ .getBoolean(ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES, false);
+ this.subnetAddresses =
+ context.getConfig().getDefaultProfile().getStringMap(ADDRESS_TRANSLATOR_SUBNET_ADDRESSES)
+ .entrySet().stream()
+ .map(
+ e -> {
+ // Quoted and/or containing forward slashes map keys in reference.conf are read to
+ // strings with additional quotes, eg. 100.64.0.0/15 -> '100.64.0."0/15"' or
+ // "100.64.0.0/15" -> '"100.64.0.0/15"'
+ String subnetCIDR = e.getKey().replaceAll("\"", "");
+ String address = e.getValue();
+ return new SubnetAddress(subnetCIDR, parseAddress(address, resolveAddresses));
+ })
+ .collect(Collectors.toList());
+ this.defaultAddress =
+ Optional.ofNullable(
+ context
+ .getConfig()
+ .getDefaultProfile()
+ .getString(ADDRESS_TRANSLATOR_DEFAULT_ADDRESS, null))
+ .map(address -> parseAddress(address, resolveAddresses));
+
+ validateSubnetsAreOfSameProtocol(this.subnetAddresses);
+ validateSubnetsAreNotOverlapping(this.subnetAddresses);
+ }
+
+ private static void validateSubnetsAreOfSameProtocol(List subnets) {
+ for (int i = 0; i < subnets.size() - 1; i++) {
+ for (int j = i + 1; j < subnets.size(); j++) {
+ SubnetAddress subnet1 = subnets.get(i);
+ SubnetAddress subnet2 = subnets.get(j);
+ if (subnet1.isIPv4() != subnet2.isIPv4() && subnet1.isIPv6() != subnet2.isIPv6()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Configured subnets are of the different protocols: %s, %s", subnet1, subnet2));
+ }
+ }
+ }
+ }
+
+ private static void validateSubnetsAreNotOverlapping(List subnets) {
+ for (int i = 0; i < subnets.size() - 1; i++) {
+ for (int j = i + 1; j < subnets.size(); j++) {
+ SubnetAddress subnet1 = subnets.get(i);
+ SubnetAddress subnet2 = subnets.get(j);
+ if (subnet1.isOverlapping(subnet2)) {
+ throw new IllegalArgumentException(
+ String.format("Configured subnets are overlapping: %s, %s", subnet1, subnet2));
+ }
+ }
+ }
+ }
+
+ @NonNull
+ @Override
+ public InetSocketAddress translate(@NonNull InetSocketAddress address) {
+ InetSocketAddress translatedAddress = null;
+ for (SubnetAddress subnetAddress : subnetAddresses) {
+ if (subnetAddress.contains(address)) {
+ translatedAddress = subnetAddress.getAddress();
+ }
+ }
+ if (translatedAddress == null && defaultAddress.isPresent()) {
+ translatedAddress = defaultAddress.get();
+ }
+ if (translatedAddress == null) {
+ translatedAddress = address;
+ }
+ LOG.debug("[{}] Translated {} to {}", logPrefix, address, translatedAddress);
+ return translatedAddress;
+ }
+
+ @Override
+ public void close() {}
+
+ @Nullable
+ private InetSocketAddress parseAddress(String address, boolean resolve) {
+ try {
+ InetSocketAddress parsedAddress = AddressUtils.extract(address, resolve).iterator().next();
+ LOG.debug("[{}] Parsed {} to {}", logPrefix, address, parsedAddress);
+ return parsedAddress;
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException(
+ String.format("Invalid address %s (%s)", address, e.getMessage()), e);
+ }
+ }
+}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java
index 7c14a811b61..3edf8f2b5c3 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java
@@ -46,6 +46,7 @@
import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy;
import com.datastax.oss.driver.api.core.ssl.SslEngineFactory;
import com.datastax.oss.driver.api.core.time.TimestampGenerator;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
import com.datastax.oss.driver.api.core.tracker.RequestTracker;
import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
@@ -229,11 +230,12 @@ public class DefaultDriverContext implements InternalDriverContext {
new LazyReference<>("metricIdGenerator", this::buildMetricIdGenerator, cycleDetector);
private final LazyReference requestThrottlerRef =
new LazyReference<>("requestThrottler", this::buildRequestThrottler, cycleDetector);
- private final LazyReference> startupOptionsRef =
- new LazyReference<>("startupOptions", this::buildStartupOptions, cycleDetector);
+ private final LazyReference startupOptionsRef =
+ new LazyReference<>("startupOptionsFactory", this::buildStartupOptionsFactory, cycleDetector);
private final LazyReference nodeStateListenerRef;
private final LazyReference schemaChangeListenerRef;
private final LazyReference requestTrackerRef;
+ private final LazyReference> requestIdGeneratorRef;
private final LazyReference> authProviderRef;
private final LazyReference> lifecycleListenersRef =
new LazyReference<>("lifecycleListeners", this::buildLifecycleListeners, cycleDetector);
@@ -296,6 +298,11 @@ public DefaultDriverContext(
this.requestTrackerRef =
new LazyReference<>(
"requestTracker", () -> buildRequestTracker(requestTrackerFromBuilder), cycleDetector);
+ this.requestIdGeneratorRef =
+ new LazyReference<>(
+ "requestIdGenerator",
+ () -> buildRequestIdGenerator(programmaticArguments.getRequestIdGenerator()),
+ cycleDetector);
this.sslEngineFactoryRef =
new LazyReference<>(
"sslEngineFactory",
@@ -350,16 +357,15 @@ public DefaultDriverContext(
}
/**
- * Builds a map of options to send in a Startup message.
+ * Returns builder of options to send in a Startup message.
*
* @see #getStartupOptions()
*/
- protected Map buildStartupOptions() {
+ protected StartupOptionsBuilder buildStartupOptionsFactory() {
return new StartupOptionsBuilder(this)
.withClientId(startupClientId)
.withApplicationName(startupApplicationName)
- .withApplicationVersion(startupApplicationVersion)
- .build();
+ .withApplicationVersion(startupApplicationVersion);
}
protected Map buildLoadBalancingPolicies() {
@@ -895,6 +901,17 @@ protected RequestTracker buildRequestTracker(RequestTracker requestTrackerFromBu
}
}
+ protected Optional buildRequestIdGenerator(
+ RequestIdGenerator requestIdGenerator) {
+ return (requestIdGenerator != null)
+ ? Optional.of(requestIdGenerator)
+ : Reflection.buildFromConfig(
+ this,
+ DefaultDriverOption.REQUEST_ID_GENERATOR_CLASS,
+ RequestIdGenerator.class,
+ "com.datastax.oss.driver.internal.core.tracker");
+ }
+
protected Optional buildAuthProvider(AuthProvider authProviderFromBuilder) {
return (authProviderFromBuilder != null)
? Optional.of(authProviderFromBuilder)
@@ -1159,6 +1176,12 @@ public RequestTracker getRequestTracker() {
return requestTrackerRef.get();
}
+ @NonNull
+ @Override
+ public Optional getRequestIdGenerator() {
+ return requestIdGeneratorRef.get();
+ }
+
@Nullable
@Override
public String getLocalDatacenter(@NonNull String profileName) {
@@ -1199,7 +1222,8 @@ public ProtocolVersion getProtocolVersion() {
@NonNull
@Override
public Map getStartupOptions() {
- return startupOptionsRef.get();
+ // startup options are calculated dynamically and may vary per connection
+ return startupOptionsRef.get().build();
}
protected RequestLogFormatter buildRequestLogFormatter() {
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultNettyOptions.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultNettyOptions.java
index c5d3b3670f0..763a71f8b12 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultNettyOptions.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultNettyOptions.java
@@ -200,7 +200,7 @@ public Future onClose() {
}
@Override
- public synchronized Timer getTimer() {
+ public Timer getTimer() {
return timer;
}
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
index 684d6b01b9c..89a9266b3ac 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
@@ -19,24 +19,34 @@
import com.datastax.dse.driver.api.core.config.DseDriverOption;
import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
+import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy;
import com.datastax.oss.driver.api.core.session.Session;
import com.datastax.oss.driver.api.core.uuid.Uuids;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
import com.datastax.oss.protocol.internal.request.Startup;
import com.datastax.oss.protocol.internal.util.collection.NullAllowingImmutableMap;
+import com.fasterxml.jackson.databind.ObjectMapper;
import edu.umd.cs.findbugs.annotations.Nullable;
import java.util.Map;
+import java.util.Optional;
import java.util.UUID;
import net.jcip.annotations.Immutable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
@Immutable
public class StartupOptionsBuilder {
public static final String DRIVER_NAME_KEY = "DRIVER_NAME";
public static final String DRIVER_VERSION_KEY = "DRIVER_VERSION";
+ public static final String DRIVER_BAGGAGE = "DRIVER_BAGGAGE";
public static final String APPLICATION_NAME_KEY = "APPLICATION_NAME";
public static final String APPLICATION_VERSION_KEY = "APPLICATION_VERSION";
public static final String CLIENT_ID_KEY = "CLIENT_ID";
+ private static final Logger LOG = LoggerFactory.getLogger(StartupOptionsBuilder.class);
+ private static final ObjectMapper mapper = new ObjectMapper();
+
protected final InternalDriverContext context;
private UUID clientId;
private String applicationName;
@@ -119,6 +129,7 @@ public Map build() {
if (applicationVersion != null) {
builder.put(APPLICATION_VERSION_KEY, applicationVersion);
}
+ driverBaggage().ifPresent(s -> builder.put(DRIVER_BAGGAGE, s));
return builder.build();
}
@@ -142,4 +153,21 @@ protected String getDriverName() {
protected String getDriverVersion() {
return Session.OSS_DRIVER_COORDINATES.getVersion().toString();
}
+
+ private Optional driverBaggage() {
+ ImmutableMap.Builder builder = new ImmutableMap.Builder<>();
+ for (Map.Entry entry :
+ context.getLoadBalancingPolicies().entrySet()) {
+ Map config = entry.getValue().getStartupConfiguration();
+ if (!config.isEmpty()) {
+ builder.put(entry.getKey(), config);
+ }
+ }
+ try {
+ return Optional.of(mapper.writeValueAsString(builder.build()));
+ } catch (Exception e) {
+ LOG.warn("Failed to construct startup driver baggage", e);
+ return Optional.empty();
+ }
+ }
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java
index d38443524f2..57411097b82 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java
@@ -162,16 +162,13 @@ public CompletionStage process(
});
}
}
+ // If the future is already completed, return it directly to maintain a strong reference
+ // in the cache and avoid premature GC with weakValues() (ScyllaDB PR #892).
if (result.isDone()) {
- // Once a CompletableFuture is completed, cancel(), complete(), and
- // completeExceptionally() are no-ops (return false), so returning the cached
- // instance directly is safe from concurrent callers. This also keeps the cache
- // entry alive via the caller's strong reference, preventing premature weak-value
- // eviction under GC pressure.
return result;
}
- // Defensive copy for in-flight preparations only: protects the shared cached future
- // from cancellation by one of multiple concurrent waiters.
+ // Return a defensive copy. So if a client cancels its request, the cache won't be impacted
+ // nor a potential concurrent request.
return result.thenApply(x -> x); // copy() is available only since Java 9
} catch (ExecutionException e) {
return CompletableFutures.failedFuture(e.getCause());
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java
index 25f27e5a39d..29f595e5225 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java
@@ -58,6 +58,7 @@
import com.datastax.oss.driver.api.core.servererrors.WriteTimeoutException;
import com.datastax.oss.driver.api.core.session.throttling.RequestThrottler;
import com.datastax.oss.driver.api.core.session.throttling.Throttled;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
import com.datastax.oss.driver.api.core.tracker.RequestTracker;
import com.datastax.oss.driver.internal.core.adminrequest.ThrottledAdminRequestHandler;
import com.datastax.oss.driver.internal.core.adminrequest.UnexpectedResponseException;
@@ -77,6 +78,7 @@
import com.datastax.oss.driver.internal.core.tracker.RequestLogger;
import com.datastax.oss.driver.internal.core.util.Loggers;
import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan;
+import com.datastax.oss.driver.shaded.guava.common.base.Joiner;
import com.datastax.oss.protocol.internal.Frame;
import com.datastax.oss.protocol.internal.Message;
import com.datastax.oss.protocol.internal.ProtocolConstants;
@@ -101,6 +103,7 @@
import java.util.AbstractMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
@@ -119,7 +122,7 @@ public class CqlRequestHandler implements Throttled {
private static final long NANOTIME_NOT_MEASURED_YET = -1;
private final long startTimeNanos;
- private final String logPrefix;
+ private final String handlerLogPrefix;
private final Statement initialStatement;
private final DefaultSession session;
private final CqlIdentifier keyspace;
@@ -144,6 +147,7 @@ public class CqlRequestHandler implements Throttled {
private final List inFlightCallbacks;
private final RequestThrottler throttler;
private final RequestTracker requestTracker;
+ private final Optional requestIdGenerator;
private final SessionMetricUpdater sessionMetricUpdater;
private final DriverExecutionProfile executionProfile;
@@ -151,15 +155,25 @@ public class CqlRequestHandler implements Throttled {
// We don't use a map because nodes can appear multiple times.
private volatile List> errors;
+ private final Joiner logPrefixJoiner = Joiner.on('|');
+ private final String sessionName;
+ private final String sessionRequestId;
+
protected CqlRequestHandler(
Statement statement,
DefaultSession session,
InternalDriverContext context,
- String sessionLogPrefix) {
+ String sessionName) {
this.startTimeNanos = System.nanoTime();
- this.logPrefix = sessionLogPrefix + "|" + this.hashCode();
- LOG.trace("[{}] Creating new handler for request {}", logPrefix, statement);
+ this.requestIdGenerator = context.getRequestIdGenerator();
+ this.sessionName = sessionName;
+ this.sessionRequestId =
+ this.requestIdGenerator
+ .map(RequestIdGenerator::getSessionRequestId)
+ .orElse(Integer.toString(this.hashCode()));
+ this.handlerLogPrefix = logPrefixJoiner.join(sessionName, sessionRequestId);
+ LOG.trace("[{}] Creating new handler for request {}", handlerLogPrefix, statement);
this.initialStatement = statement;
this.session = session;
@@ -174,7 +188,7 @@ protected CqlRequestHandler(
context.getRequestThrottler().signalCancel(this);
}
} catch (Throwable t2) {
- Loggers.warnWithException(LOG, "[{}] Uncaught exception", logPrefix, t2);
+ Loggers.warnWithException(LOG, "[{}] Uncaught exception", handlerLogPrefix, t2);
}
return null;
});
@@ -370,7 +384,7 @@ private void sendRequest(
|| (channel =
session.getChannel(
node,
- logPrefix,
+ handlerLogPrefix,
getRoutingToken(statement),
getShardFromTabletMap(statement, node, getRoutingToken(statement))))
== null) {
@@ -378,7 +392,7 @@ private void sendRequest(
channel =
session.getChannel(
node,
- logPrefix,
+ handlerLogPrefix,
getRoutingToken(statement),
getShardFromTabletMap(statement, node, getRoutingToken(statement)));
if (channel != null) {
@@ -395,6 +409,16 @@ private void sendRequest(
setFinalError(statement, AllNodesFailedException.fromErrors(this.errors), null, -1);
}
} else {
+ Statement finalStatement = statement;
+ String nodeRequestId =
+ this.requestIdGenerator
+ .map((g) -> g.getNodeRequestId(finalStatement, sessionRequestId))
+ .orElse(Integer.toString(this.hashCode()));
+ statement =
+ this.requestIdGenerator
+ .map((g) -> g.getDecoratedStatement(finalStatement, nodeRequestId))
+ .orElse(finalStatement);
+
NodeResponseCallback nodeResponseCallback =
new NodeResponseCallback(
statement,
@@ -404,7 +428,7 @@ private void sendRequest(
currentExecutionIndex,
retryCount,
scheduleNextExecution,
- logPrefix);
+ logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex));
Message message = Conversions.toMessage(statement, executionProfile, context);
channel
.write(message, statement.isTracing(), statement.getCustomPayload(), nodeResponseCallback)
@@ -463,9 +487,17 @@ private void setFinalResult(
totalLatencyNanos = completionTimeNanos - startTimeNanos;
long nodeLatencyNanos = completionTimeNanos - callback.nodeStartTimeNanos;
requestTracker.onNodeSuccess(
- callback.statement, nodeLatencyNanos, executionProfile, callback.node, logPrefix);
+ callback.statement,
+ nodeLatencyNanos,
+ executionProfile,
+ callback.node,
+ handlerLogPrefix);
requestTracker.onSuccess(
- callback.statement, totalLatencyNanos, executionProfile, callback.node, logPrefix);
+ callback.statement,
+ totalLatencyNanos,
+ executionProfile,
+ callback.node,
+ handlerLogPrefix);
}
if (sessionMetricUpdater.isEnabled(
DefaultSessionMetric.CQL_REQUESTS, executionProfile.getName())) {
@@ -589,7 +621,8 @@ private void setFinalError(Statement statement, Throwable error, Node node, i
cancelScheduledTasks();
if (!(requestTracker instanceof NoopRequestTracker)) {
long latencyNanos = System.nanoTime() - startTimeNanos;
- requestTracker.onError(statement, error, latencyNanos, executionProfile, node, logPrefix);
+ requestTracker.onError(
+ statement, error, latencyNanos, executionProfile, node, handlerLogPrefix);
}
if (error instanceof DriverTimeoutException) {
throttler.signalTimeout(this);
@@ -639,7 +672,7 @@ private NodeResponseCallback(
this.execution = execution;
this.retryCount = retryCount;
this.scheduleNextExecution = scheduleNextExecution;
- this.logPrefix = logPrefix + "|" + execution;
+ this.logPrefix = logPrefix;
}
// this gets invoked once the write completes.
@@ -717,7 +750,7 @@ private void scheduleSpeculativeExecution(int index, long delay) {
if (!result.isDone()) {
LOG.trace(
"[{}] Starting speculative execution {}",
- CqlRequestHandler.this.logPrefix,
+ CqlRequestHandler.this.handlerLogPrefix,
index);
activeExecutionsCount.incrementAndGet();
startedSpeculativeExecutionsCount.incrementAndGet();
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java
index f0e1cb3cf45..32eaad4973d 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java
@@ -59,6 +59,7 @@
import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan;
import com.datastax.oss.driver.shaded.guava.common.base.Predicates;
import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
import com.datastax.oss.driver.shaded.guava.common.collect.Lists;
import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
import edu.umd.cs.findbugs.annotations.NonNull;
@@ -173,6 +174,9 @@ public BasicLoadBalancingPolicy(@NonNull DriverContext context, @NonNull String
private RequestRoutingMethod parseLwtRequestRoutingMethod() {
String methodString =
profile.getString(DefaultDriverOption.LOAD_BALANCING_DEFAULT_LWT_REQUEST_ROUTING_METHOD);
+ if (methodString == null) {
+ return RequestRoutingMethod.PRESERVE_REPLICA_ORDER;
+ }
try {
return RequestRoutingMethod.valueOf(methodString.toUpperCase());
} catch (IllegalArgumentException e) {
@@ -232,7 +236,7 @@ private Optional getRequestProfile(@NonNull Request requ
* Before initialization, this method always returns null.
*/
@Nullable
- protected String getLocalDatacenter() {
+ public String getLocalDatacenter() {
return localDc;
}
@@ -241,6 +245,34 @@ protected String getLocalRack() {
return localRack;
}
+ @NonNull
+ @Override
+ public Map getStartupConfiguration() {
+ ImmutableMap.Builder builder = new ImmutableMap.Builder<>();
+ if (localDc != null) {
+ builder.put("localDc", localDc);
+ } else {
+ // Local data center may not be discovered prior to connection pool initialization.
+ // In such scenario, return configured local data center name.
+ // Note that when using DC inferring load balancing policy, startup configuration
+ // may not show local DC name, because it will be discovered only once control connection
+ // is established and datacenter of contact points known.
+ Optional configuredDc =
+ new OptionalLocalDcHelper(context, profile, logPrefix).configuredLocalDc();
+ configuredDc.ifPresent(d -> builder.put("localDc", d));
+ }
+ if (!preferredRemoteDcs.isEmpty()) {
+ builder.put("preferredRemoteDcs", preferredRemoteDcs);
+ }
+ if (allowDcFailoverForLocalCl) {
+ builder.put("allowDcFailoverForLocalCl", allowDcFailoverForLocalCl);
+ }
+ if (maxNodesPerRemoteDc > 0) {
+ builder.put("maxNodesPerRemoteDc", maxNodesPerRemoteDc);
+ }
+ return ImmutableMap.of(BasicLoadBalancingPolicy.class.getSimpleName(), builder.build());
+ }
+
/** @return The nodes currently considered as live. */
protected NodeSet getLiveNodes() {
return liveNodes;
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java
index 6b0935a887d..fd7cf812142 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java
@@ -34,6 +34,7 @@
import com.datastax.oss.driver.internal.core.util.collection.QueryPlan;
import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan;
import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
import com.datastax.oss.driver.shaded.guava.common.collect.MapMaker;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.Nullable;
@@ -267,7 +268,7 @@ public void onNodeSuccess(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
updateResponseTimes(node);
}
@@ -278,7 +279,7 @@ public void onNodeError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
updateResponseTimes(node);
}
@@ -326,7 +327,8 @@ protected class NodeResponseRateSample {
@VisibleForTesting protected final OptionalLong newest;
private NodeResponseRateSample() {
- this.oldest = nanoTime();
+ long now = nanoTime();
+ this.oldest = now;
this.newest = OptionalLong.empty();
}
@@ -366,4 +368,13 @@ private boolean hasSufficientResponses(long now) {
return this.oldest - threshold >= 0;
}
}
+
+ @NonNull
+ @Override
+ public Map getStartupConfiguration() {
+ Map parent = super.getStartupConfiguration();
+ return ImmutableMap.of(
+ DefaultLoadBalancingPolicy.class.getSimpleName(),
+ parent.get(BasicLoadBalancingPolicy.class.getSimpleName()));
+ }
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java
index 97aab92fff7..fe6dcc96754 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java
@@ -26,6 +26,7 @@
import edu.umd.cs.findbugs.annotations.NonNull;
import java.util.ArrayList;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -67,34 +68,62 @@ public OptionalLocalDcHelper(
@Override
@NonNull
public Optional discoverLocalDc(@NonNull Map nodes) {
- String localDc = context.getLocalDatacenter(profile.getName());
- if (localDc != null) {
- LOG.debug("[{}] Local DC set programmatically: {}", logPrefix, localDc);
- } else if (profile.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) {
- localDc = profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER);
- LOG.debug("[{}] Local DC set from configuration: {}", logPrefix, localDc);
+ Optional localDc = configuredLocalDc();
+ if (localDc.isPresent()) {
+ checkLocalDatacenterCompatibility(
+ localDc.get(), context.getMetadataManager().getContactPoints());
+ // Also warn if the configured DC doesn't match any node in the cluster
+ if (!nodes.isEmpty()) {
+ boolean found = false;
+ for (Node node : nodes.values()) {
+ if (localDc.get().equals(node.getDatacenter())) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ LOG.warn(
+ "[{}] Configured local DC '{}' does not match any node's datacenter"
+ + " (available DCs: {}); please verify your configuration",
+ logPrefix,
+ localDc.get(),
+ formatDcs(nodes.values()));
+ }
+ }
} else {
LOG.debug("[{}] Local DC not set, DC awareness will be disabled", logPrefix);
- return Optional.empty();
}
- if (!nodes.isEmpty()) {
- boolean found = false;
- for (Node node : nodes.values()) {
- if (localDc.equals(node.getDatacenter())) {
- found = true;
- break;
+ return localDc;
+ }
+
+ /**
+ * Checks if the contact points are compatible with the local datacenter specified either through
+ * configuration, or programmatically.
+ *
+ * The default implementation logs a warning when a contact point reports a datacenter
+ * different from the local one, and only for the default profile.
+ *
+ * @param localDc The local datacenter, as specified in the config, or programmatically.
+ * @param contactPoints The contact points provided when creating the session.
+ */
+ protected void checkLocalDatacenterCompatibility(
+ @NonNull String localDc, Set contactPoints) {
+ if (profile.getName().equals(DriverExecutionProfile.DEFAULT_NAME)) {
+ Set badContactPoints = new LinkedHashSet<>();
+ for (Node node : contactPoints) {
+ if (!Objects.equals(localDc, node.getDatacenter())) {
+ badContactPoints.add(node);
}
}
- if (!found) {
+ if (!badContactPoints.isEmpty()) {
LOG.warn(
- "[{}] Configured local DC '{}' does not match any node's datacenter"
- + " (available DCs: {}); please verify your configuration",
+ "[{}] You specified {} as the local DC, but some contact points are from a different DC: {}; "
+ + "please provide the correct local DC, or check your contact points",
logPrefix,
localDc,
- formatDcs(nodes.values()));
+ formatNodesAndDcs(badContactPoints));
}
}
- return Optional.of(localDc);
}
/**
@@ -166,4 +195,19 @@ protected String formatDcs(Iterable nodes) {
}
return String.join(", ", new TreeSet<>(l));
}
+
+ /** @return Local data center set programmatically or from configuration file. */
+ @NonNull
+ public Optional configuredLocalDc() {
+ String localDc = context.getLocalDatacenter(profile.getName());
+ if (localDc != null) {
+ LOG.debug("[{}] Local DC set programmatically: {}", logPrefix, localDc);
+ return Optional.of(localDc);
+ } else if (profile.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) {
+ localDc = profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER);
+ LOG.debug("[{}] Local DC set from configuration: {}", logPrefix, localDc);
+ return Optional.of(localDc);
+ }
+ return Optional.empty();
+ }
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java
index ace4e82617d..d1ab8eec98d 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java
@@ -26,10 +26,10 @@
import java.util.Arrays;
import java.util.Comparator;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicInteger;
public class SniEndPoint implements EndPoint {
- private static final AtomicLong OFFSET = new AtomicLong();
+ private static final AtomicInteger OFFSET = new AtomicInteger();
private final InetSocketAddress proxyAddress;
private final String serverName;
@@ -64,7 +64,10 @@ public InetSocketAddress resolve() {
// The order of the returned address is unspecified. Sort by IP to make sure we get a true
// round-robin
Arrays.sort(aRecords, IP_COMPARATOR);
- int index = (aRecords.length == 1) ? 0 : (int) OFFSET.getAndIncrement() % aRecords.length;
+ int index =
+ (aRecords.length == 1)
+ ? 0
+ : OFFSET.getAndUpdate(x -> x == Integer.MAX_VALUE ? 0 : x + 1) % aRecords.length;
return new InetSocketAddress(aRecords[index], proxyAddress.getPort());
} catch (UnknownHostException e) {
throw new IllegalArgumentException(
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottler.java
index ffe0ffe9650..8146c5b113a 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottler.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottler.java
@@ -26,10 +26,9 @@
import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.Nullable;
-import java.util.ArrayDeque;
import java.util.Deque;
-import java.util.concurrent.locks.ReentrantLock;
-import net.jcip.annotations.GuardedBy;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicInteger;
import net.jcip.annotations.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,17 +60,12 @@ public class ConcurrencyLimitingRequestThrottler implements RequestThrottler {
private final String logPrefix;
private final int maxConcurrentRequests;
private final int maxQueueSize;
-
- private final ReentrantLock lock = new ReentrantLock();
-
- @GuardedBy("lock")
- private int concurrentRequests;
-
- @GuardedBy("lock")
- private final Deque queue = new ArrayDeque<>();
-
- @GuardedBy("lock")
- private boolean closed;
+ private final AtomicInteger concurrentRequests = new AtomicInteger(0);
+ // CLQ is not O(1) for size(), as it forces a full iteration of the queue. So, we track
+ // the size of the queue explicitly.
+ private final Deque queue = new ConcurrentLinkedDeque<>();
+ private final AtomicInteger queueSize = new AtomicInteger(0);
+ private volatile boolean closed = false;
public ConcurrencyLimitingRequestThrottler(DriverContext context) {
this.logPrefix = context.getSessionName();
@@ -88,50 +82,62 @@ public ConcurrencyLimitingRequestThrottler(DriverContext context) {
@Override
public void register(@NonNull Throttled request) {
- boolean notifyReadyRequired = false;
+ if (closed) {
+ LOG.trace("[{}] Rejecting request after shutdown", logPrefix);
+ fail(request, "The session is shutting down");
+ return;
+ }
- lock.lock();
- try {
- if (closed) {
- LOG.trace("[{}] Rejecting request after shutdown", logPrefix);
- fail(request, "The session is shutting down");
- } else if (queue.isEmpty() && concurrentRequests < maxConcurrentRequests) {
- // We have capacity for one more concurrent request
+ // Implementation note: Technically the "concurrent requests" or "queue size"
+ // could read transiently over the limit, but the queue itself will never grow
+ // beyond the limit since we always check for that condition and revert if
+ // over-limit. We do this instead of a CAS-loop to avoid the potential loop.
+
+ // If no backlog exists AND we get capacity, we can execute immediately
+ if (queueSize.get() == 0) {
+ // Take a claim first, and then check if we are OK to proceed
+ int newConcurrent = concurrentRequests.incrementAndGet();
+ if (newConcurrent <= maxConcurrentRequests) {
LOG.trace("[{}] Starting newly registered request", logPrefix);
- concurrentRequests += 1;
- notifyReadyRequired = true;
- } else if (queue.size() < maxQueueSize) {
- LOG.trace("[{}] Enqueuing request", logPrefix);
- queue.add(request);
+ request.onThrottleReady(false);
+ return;
} else {
- LOG.trace("[{}] Rejecting request because of full queue", logPrefix);
- fail(
- request,
- String.format(
- "The session has reached its maximum capacity "
- + "(concurrent requests: %d, queue size: %d)",
- maxConcurrentRequests, maxQueueSize));
+ // We exceeded the limit, decrement the count and fall through to the queuing logic
+ concurrentRequests.decrementAndGet();
}
- } finally {
- lock.unlock();
}
- // no need to hold the lock while allowing the task to progress
- if (notifyReadyRequired) {
- request.onThrottleReady(false);
+ // If we have a backlog, or we failed to claim capacity, try to enqueue
+ int newQueueSize = queueSize.incrementAndGet();
+ if (newQueueSize <= maxQueueSize) {
+ LOG.trace("[{}] Enqueuing request", logPrefix);
+ queue.offer(request);
+
+ // Double-check that we were still supposed to be enqueued; it is possible
+ // that the session was closed while we were enqueuing, it's also possible
+ // that it is right now removing the request, so we need to check both
+ if (closed) {
+ if (queue.remove(request)) {
+ queueSize.decrementAndGet();
+ LOG.trace("[{}] Rejecting late request after shutdown", logPrefix);
+ fail(request, "The session is shutting down");
+ }
+ }
+ } else {
+ LOG.trace("[{}] Rejecting request because of full queue", logPrefix);
+ queueSize.decrementAndGet();
+ fail(
+ request,
+ String.format(
+ "The session has reached its maximum capacity "
+ + "(concurrent requests: %d, queue size: %d)",
+ maxConcurrentRequests, maxQueueSize));
}
}
@Override
public void signalSuccess(@NonNull Throttled request) {
- Throttled nextRequest = null;
- lock.lock();
- try {
- nextRequest = onRequestDoneAndDequeNext();
- } finally {
- lock.unlock();
- }
-
+ Throttled nextRequest = onRequestDoneAndDequeNext();
if (nextRequest != null) {
nextRequest.onThrottleReady(true);
}
@@ -145,17 +151,13 @@ public void signalError(@NonNull Throttled request, @NonNull Throwable error) {
@Override
public void signalTimeout(@NonNull Throttled request) {
Throttled nextRequest = null;
- lock.lock();
- try {
- if (!closed) {
- if (queue.remove(request)) { // The request timed out before it was active
- LOG.trace("[{}] Removing timed out request from the queue", logPrefix);
- } else {
- nextRequest = onRequestDoneAndDequeNext();
- }
+ if (!closed) {
+ if (queue.remove(request)) { // The request timed out before it was active
+ queueSize.decrementAndGet();
+ LOG.trace("[{}] Removing timed out request from the queue", logPrefix);
+ } else {
+ nextRequest = onRequestDoneAndDequeNext();
}
- } finally {
- lock.unlock();
}
if (nextRequest != null) {
@@ -166,17 +168,13 @@ public void signalTimeout(@NonNull Throttled request) {
@Override
public void signalCancel(@NonNull Throttled request) {
Throttled nextRequest = null;
- lock.lock();
- try {
- if (!closed) {
- if (queue.remove(request)) { // The request has been cancelled before it was active
- LOG.trace("[{}] Removing cancelled request from the queue", logPrefix);
- } else {
- nextRequest = onRequestDoneAndDequeNext();
- }
+ if (!closed) {
+ if (queue.remove(request)) { // The request has been cancelled before it was active
+ queueSize.decrementAndGet();
+ LOG.trace("[{}] Removing cancelled request from the queue", logPrefix);
+ } else {
+ nextRequest = onRequestDoneAndDequeNext();
}
- } finally {
- lock.unlock();
}
if (nextRequest != null) {
@@ -184,17 +182,16 @@ public void signalCancel(@NonNull Throttled request) {
}
}
- @SuppressWarnings("GuardedBy") // this method is only called with the lock held
@Nullable
private Throttled onRequestDoneAndDequeNext() {
- assert lock.isHeldByCurrentThread();
if (!closed) {
- if (queue.isEmpty()) {
- concurrentRequests -= 1;
+ Throttled nextRequest = queue.poll();
+ if (nextRequest == null) {
+ concurrentRequests.decrementAndGet();
} else {
+ queueSize.decrementAndGet();
LOG.trace("[{}] Starting dequeued request", logPrefix);
- // don't touch concurrentRequests since we finished one but started another
- return queue.poll();
+ return nextRequest;
}
}
@@ -204,45 +201,28 @@ private Throttled onRequestDoneAndDequeNext() {
@Override
public void close() {
- lock.lock();
- try {
- closed = true;
- LOG.debug("[{}] Rejecting {} queued requests after shutdown", logPrefix, queue.size());
- for (Throttled request : queue) {
- fail(request, "The session is shutting down");
- }
- } finally {
- lock.unlock();
+ closed = true;
+
+ LOG.debug("[{}] Rejecting {} queued requests after shutdown", logPrefix, queueSize.get());
+ Throttled request;
+ while ((request = queue.poll()) != null) {
+ queueSize.decrementAndGet();
+ fail(request, "The session is shutting down");
}
}
public int getQueueSize() {
- lock.lock();
- try {
- return queue.size();
- } finally {
- lock.unlock();
- }
+ return queueSize.get();
}
@VisibleForTesting
int getConcurrentRequests() {
- lock.lock();
- try {
- return concurrentRequests;
- } finally {
- lock.unlock();
- }
+ return concurrentRequests.get();
}
@VisibleForTesting
Deque getQueue() {
- lock.lock();
- try {
- return queue;
- } finally {
- lock.unlock();
- }
+ return queue;
}
private static void fail(Throttled request, String message) {
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java
index 475ec38d578..343d3f9e4e7 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java
@@ -22,6 +22,7 @@
import com.datastax.oss.driver.api.core.context.DriverContext;
import com.datastax.oss.driver.api.core.metadata.EndPoint;
import com.datastax.oss.driver.api.core.ssl.SslEngineFactory;
+import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
import edu.umd.cs.findbugs.annotations.NonNull;
import java.io.InputStream;
import java.net.InetSocketAddress;
@@ -69,6 +70,7 @@ public class DefaultSslEngineFactory implements SslEngineFactory {
private final SSLContext sslContext;
private final String[] cipherSuites;
private final boolean requireHostnameValidation;
+ private final boolean allowDnsReverseLookupSan;
private ReloadingKeyManagerFactory kmf;
/** Builds a new instance from the driver configuration. */
@@ -88,6 +90,28 @@ public DefaultSslEngineFactory(DriverContext driverContext) {
}
this.requireHostnameValidation =
config.getBoolean(DefaultDriverOption.SSL_HOSTNAME_VALIDATION, true);
+ this.allowDnsReverseLookupSan =
+ config.getBoolean(DefaultDriverOption.SSL_ALLOW_DNS_REVERSE_LOOKUP_SAN, true);
+ }
+
+ @VisibleForTesting
+ protected String hostname(InetSocketAddress addr) {
+ return allowDnsReverseLookupSan ? hostMaybeFromDnsReverseLookup(addr) : hostNoLookup(addr);
+ }
+
+ @VisibleForTesting
+ protected String hostMaybeFromDnsReverseLookup(InetSocketAddress addr) {
+ // See java.net.InetSocketAddress.getHostName:
+ // "This method may trigger a name service reverse lookup if the address was created with a
+ // literal IP address."
+ return addr.getHostName();
+ }
+
+ @VisibleForTesting
+ protected String hostNoLookup(InetSocketAddress addr) {
+ // See java.net.InetSocketAddress.getHostString:
+ // "This has the benefit of not attempting a reverse lookup"
+ return addr.getHostString();
}
@NonNull
@@ -97,7 +121,7 @@ public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) {
SocketAddress remoteAddress = remoteEndpoint.resolve();
if (remoteAddress instanceof InetSocketAddress) {
InetSocketAddress socketAddress = (InetSocketAddress) remoteAddress;
- engine = sslContext.createSSLEngine(socketAddress.getHostName(), socketAddress.getPort());
+ engine = sslContext.createSSLEngine(hostname(socketAddress), socketAddress.getPort());
} else {
engine = sslContext.createSSLEngine();
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java
index 98af19045dc..4d2cb69fbfc 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java
@@ -38,9 +38,15 @@ public class SniSslEngineFactory implements SslEngineFactory {
private final SSLContext sslContext;
private final CopyOnWriteArrayList fakePorts = new CopyOnWriteArrayList<>();
+ private final boolean allowDnsReverseLookupSan;
public SniSslEngineFactory(SSLContext sslContext) {
+ this(sslContext, true);
+ }
+
+ public SniSslEngineFactory(SSLContext sslContext, boolean allowDnsReverseLookupSan) {
this.sslContext = sslContext;
+ this.allowDnsReverseLookupSan = allowDnsReverseLookupSan;
}
@NonNull
@@ -71,8 +77,8 @@ public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) {
// To avoid that, we create a unique "fake" port for every node. We still get session reuse for
// a given node, but not across nodes. This is safe because the advisory port is only used for
// session caching.
- SSLEngine engine =
- sslContext.createSSLEngine(address.getHostName(), getFakePort(sniServerName));
+ String peerHost = allowDnsReverseLookupSan ? address.getHostName() : address.getHostString();
+ SSLEngine engine = sslContext.createSSLEngine(peerHost, getFakePort(sniServerName));
engine.setUseClientMode(true);
SSLParameters parameters = engine.getSSLParameters();
parameters.setServerNames(ImmutableList.of(new SNIHostName(sniServerName)));
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/MultiplexingRequestTracker.java b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/MultiplexingRequestTracker.java
index d4d20f3eb78..6fe2ba059bd 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/MultiplexingRequestTracker.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/MultiplexingRequestTracker.java
@@ -82,10 +82,12 @@ public void onSuccess(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String sessionRequestLogPrefix) {
invokeTrackers(
- tracker -> tracker.onSuccess(request, latencyNanos, executionProfile, node, logPrefix),
- logPrefix,
+ tracker ->
+ tracker.onSuccess(
+ request, latencyNanos, executionProfile, node, sessionRequestLogPrefix),
+ sessionRequestLogPrefix,
"onSuccess");
}
@@ -96,10 +98,12 @@ public void onError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@Nullable Node node,
- @NonNull String logPrefix) {
+ @NonNull String sessionRequestLogPrefix) {
invokeTrackers(
- tracker -> tracker.onError(request, error, latencyNanos, executionProfile, node, logPrefix),
- logPrefix,
+ tracker ->
+ tracker.onError(
+ request, error, latencyNanos, executionProfile, node, sessionRequestLogPrefix),
+ sessionRequestLogPrefix,
"onError");
}
@@ -109,10 +113,12 @@ public void onNodeSuccess(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
invokeTrackers(
- tracker -> tracker.onNodeSuccess(request, latencyNanos, executionProfile, node, logPrefix),
- logPrefix,
+ tracker ->
+ tracker.onNodeSuccess(
+ request, latencyNanos, executionProfile, node, nodeRequestLogPrefix),
+ nodeRequestLogPrefix,
"onNodeSuccess");
}
@@ -123,11 +129,12 @@ public void onNodeError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
invokeTrackers(
tracker ->
- tracker.onNodeError(request, error, latencyNanos, executionProfile, node, logPrefix),
- logPrefix,
+ tracker.onNodeError(
+ request, error, latencyNanos, executionProfile, node, nodeRequestLogPrefix),
+ nodeRequestLogPrefix,
"onNodeError");
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/NoopRequestTracker.java b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/NoopRequestTracker.java
index 09ac27e5e75..3821c6ace2d 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/NoopRequestTracker.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/NoopRequestTracker.java
@@ -42,7 +42,7 @@ public void onSuccess(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String requestPrefix) {
+ @NonNull String sessionRequestLogPrefix) {
// nothing to do
}
@@ -53,7 +53,7 @@ public void onError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
Node node,
- @NonNull String requestPrefix) {
+ @NonNull String sessionRequestLogPrefix) {
// nothing to do
}
@@ -64,7 +64,7 @@ public void onNodeError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String requestPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
// nothing to do
}
@@ -74,7 +74,7 @@ public void onNodeSuccess(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String requestPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
// nothing to do
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/RequestLogger.java b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/RequestLogger.java
index 235ef051b40..f242ff89c54 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/RequestLogger.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/RequestLogger.java
@@ -86,7 +86,7 @@ public void onSuccess(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String sessionRequestLogPrefix) {
boolean successEnabled =
executionProfile.getBoolean(DefaultDriverOption.REQUEST_LOGGER_SUCCESS_ENABLED, false);
@@ -129,7 +129,7 @@ public void onSuccess(
showValues,
maxValues,
maxValueLength,
- logPrefix);
+ sessionRequestLogPrefix);
}
@Override
@@ -139,7 +139,7 @@ public void onError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
Node node,
- @NonNull String logPrefix) {
+ @NonNull String sessionRequestLogPrefix) {
if (!executionProfile.getBoolean(DefaultDriverOption.REQUEST_LOGGER_ERROR_ENABLED, false)) {
return;
@@ -173,7 +173,7 @@ public void onError(
maxValues,
maxValueLength,
showStackTraces,
- logPrefix);
+ sessionRequestLogPrefix);
}
@Override
@@ -183,7 +183,7 @@ public void onNodeError(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
// Nothing to do
}
@@ -193,7 +193,7 @@ public void onNodeSuccess(
long latencyNanos,
@NonNull DriverExecutionProfile executionProfile,
@NonNull Node node,
- @NonNull String logPrefix) {
+ @NonNull String nodeRequestLogPrefix) {
// Nothing to do
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/UuidRequestIdGenerator.java b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/UuidRequestIdGenerator.java
new file mode 100644
index 00000000000..cc07d6717f4
--- /dev/null
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/UuidRequestIdGenerator.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.tracker;
+
+import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.api.core.session.Request;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
+import com.datastax.oss.driver.api.core.uuid.Uuids;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+public class UuidRequestIdGenerator implements RequestIdGenerator {
+ public UuidRequestIdGenerator(DriverContext context) {}
+
+ /** Generates a random v4 UUID. */
+ @Override
+ public String getSessionRequestId() {
+ return Uuids.random().toString();
+ }
+
+ /**
+ * {session-request-id}-{random-uuid} All node requests for a session request will have the same
+ * session request id
+ */
+ @Override
+ public String getNodeRequestId(@NonNull Request statement, @NonNull String parentId) {
+ return parentId + "-" + Uuids.random();
+ }
+}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/W3CContextRequestIdGenerator.java b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/W3CContextRequestIdGenerator.java
new file mode 100644
index 00000000000..fe15b93bc8e
--- /dev/null
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/tracker/W3CContextRequestIdGenerator.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.tracker;
+
+import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.api.core.session.Request;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
+import com.datastax.oss.driver.shaded.guava.common.io.BaseEncoding;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.security.SecureRandom;
+import java.util.Random;
+
+public class W3CContextRequestIdGenerator implements RequestIdGenerator {
+
+ private final Random random = new SecureRandom();
+ private final BaseEncoding baseEncoding = BaseEncoding.base16().lowerCase();
+ private final String payloadKey;
+
+ public W3CContextRequestIdGenerator(DriverContext context) {
+ payloadKey = RequestIdGenerator.super.getCustomPayloadKey();
+ }
+
+ public W3CContextRequestIdGenerator(String payloadKey) {
+ this.payloadKey = payloadKey;
+ }
+
+ /** Random 16 bytes, e.g. "4bf92f3577b34da6a3ce929d0e0e4736" */
+ @Override
+ public String getSessionRequestId() {
+ byte[] bytes = new byte[16];
+ random.nextBytes(bytes);
+ return baseEncoding.encode(bytes);
+ }
+
+ /**
+ * Following the format of W3C "traceparent" spec,
+ * https://www.w3.org/TR/trace-context/#traceparent-header-field-values e.g.
+ * "00-4bf92f3577b34da6a3ce929d0e0e4736-a3ce929d0e0e4736-01" All node requests in the same session
+ * request share the same "trace-id" field value
+ */
+ @Override
+ public String getNodeRequestId(@NonNull Request statement, @NonNull String parentId) {
+ byte[] bytes = new byte[8];
+ random.nextBytes(bytes);
+ return String.format("00-%s-%s-00", parentId, baseEncoding.encode(bytes));
+ }
+
+ @Override
+ public String getCustomPayloadKey() {
+ return this.payloadKey;
+ }
+}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/registry/DefaultCodecRegistry.java b/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/registry/DefaultCodecRegistry.java
index cfd053ea56e..cc14740e180 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/registry/DefaultCodecRegistry.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/registry/DefaultCodecRegistry.java
@@ -159,7 +159,10 @@ public boolean equals(Object other) {
@Override
public int hashCode() {
- return Objects.hash(cqlType, javaType, isJavaCovariant);
+ // NOTE: inlined Objects.hash for performance reasons (avoid Object[] allocation
+ // seen in profiler allocation traces)
+ return ((31 + Objects.hashCode(cqlType)) * 31 + Objects.hashCode(javaType)) * 31
+ + Boolean.hashCode(isJavaCovariant);
}
}
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/util/AddressUtils.java b/core/src/main/java/com/datastax/oss/driver/internal/core/util/AddressUtils.java
new file mode 100644
index 00000000000..8905edb9192
--- /dev/null
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/util/AddressUtils.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.util;
+
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.HashSet;
+import java.util.Set;
+
+public class AddressUtils {
+
+ public static Set extract(String address, boolean resolve) {
+ int separator = address.lastIndexOf(':');
+ if (separator < 0) {
+ throw new IllegalArgumentException("expecting format host:port");
+ }
+
+ String host = address.substring(0, separator);
+ String portString = address.substring(separator + 1);
+ int port;
+ try {
+ port = Integer.parseInt(portString);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("expecting port to be a number, got " + portString, e);
+ }
+ if (!resolve) {
+ return ImmutableSet.of(InetSocketAddress.createUnresolved(host, port));
+ } else {
+ InetAddress[] inetAddresses;
+ try {
+ inetAddresses = InetAddress.getAllByName(host);
+ } catch (UnknownHostException e) {
+ throw new RuntimeException(e);
+ }
+ Set result = new HashSet<>();
+ for (InetAddress inetAddress : inetAddresses) {
+ result.add(new InetSocketAddress(inetAddress, port));
+ }
+ return result;
+ }
+ }
+}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/util/ArrayUtils.java b/core/src/main/java/com/datastax/oss/driver/internal/core/util/ArrayUtils.java
index 17b6c01178d..8862c1cd6c4 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/util/ArrayUtils.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/util/ArrayUtils.java
@@ -18,6 +18,7 @@
package com.datastax.oss.driver.internal.core.util;
import edu.umd.cs.findbugs.annotations.NonNull;
+import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
public class ArrayUtils {
@@ -77,7 +78,7 @@ public static void shuffleHead(@NonNull ElementT[] elements, int n) {
* Fisher-Yates shuffle
*/
public static void shuffleHead(
- @NonNull ElementT[] elements, int n, @NonNull ThreadLocalRandom random) {
+ @NonNull ElementT[] elements, int n, @NonNull Random random) {
if (n > elements.length) {
throw new ArrayIndexOutOfBoundsException(
String.format(
diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf
index 92c5b98118f..721afa1cd76 100644
--- a/core/src/main/resources/reference.conf
+++ b/core/src/main/resources/reference.conf
@@ -866,6 +866,12 @@ datastax-java-driver {
# name matches the hostname of the server being connected to. If not set, defaults to true.
// hostname-validation = true
+ # Whether or not to allow a DNS reverse-lookup of provided server addresses for SAN addresses,
+ # if cluster endpoints are specified as literal IPs.
+ # This is left as true for compatibility, but in most environments a DNS reverse-lookup should
+ # not be necessary to get an address that matches the server certificate SANs.
+ // allow-dns-reverse-lookup-san = true
+
# The locations and passwords used to access truststore and keystore contents.
# These properties are optional. If either truststore-path or keystore-path are specified,
# the driver builds an SSLContext from these files. If neither option is specified, the
@@ -989,6 +995,13 @@ datastax-java-driver {
}
}
+ advanced.request-id {
+ generator {
+ # The component that generates a unique identifier for each CQL request, and possibly write the id to the custom payload .
+ // class = W3CContextRequestIdGenerator
+ }
+ }
+
# A session-wide component that controls the rate at which requests are executed.
#
# Implementations vary, but throttlers generally track a metric that represents the level of
@@ -1097,8 +1110,9 @@ datastax-java-driver {
# the package com.datastax.oss.driver.internal.core.addresstranslation.
#
# The driver provides the following implementations out of the box:
- # - PassThroughAddressTranslator: returns all addresses unchanged
+ # - PassThroughAddressTranslator: returns all addresses unchanged.
# - FixedHostNameAddressTranslator: translates all addresses to a specific hostname.
+ # - SubnetAddressTranslator: translates addresses to hostname based on the subnet match.
# - Ec2MultiRegionAddressTranslator: suitable for an Amazon multi-region EC2 deployment where
# clients are also deployed in EC2. It optimizes network costs by favoring private IPs over
# public ones whenever possible.
@@ -1106,8 +1120,23 @@ datastax-java-driver {
# You can also specify a custom class that implements AddressTranslator and has a public
# constructor with a DriverContext argument.
class = PassThroughAddressTranslator
+ #
# This property has to be set only in case you use FixedHostNameAddressTranslator.
# advertised-hostname = mycustomhostname
+ #
+ # These properties are only applicable in case you use SubnetAddressTranslator.
+ # subnet-addresses {
+ # "100.64.0.0/15" = "cassandra.datacenter1.com:9042"
+ # "100.66.0.0/15" = "cassandra.datacenter2.com:9042"
+ # # IPv6 example:
+ # # "::ffff:6440:0/111" = "cassandra.datacenter1.com:9042"
+ # # "::ffff:6442:0/111" = "cassandra.datacenter2.com:9042"
+ # }
+ # Optional. When configured, addresses not matching the configured subnets are translated to this address.
+ # default-address = "cassandra.datacenter1.com:9042"
+ # Whether to resolve the addresses once on initialization (if true) or on each node (re-)connection (if false).
+ # If not configured, defaults to false.
+ # resolve-addresses = false
}
# Client routes configuration for cloud private-endpoint deployments
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/ContactPointsTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/ContactPointsTest.java
index 9e0d8737619..72b875b8602 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/ContactPointsTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/ContactPointsTest.java
@@ -121,7 +121,7 @@ public void should_ignore_malformed_host_and_port_and_warn() {
ContactPoints.merge(Collections.emptySet(), ImmutableList.of("foobar"), true);
assertThat(endPoints).isEmpty();
- assertLog(Level.WARN, "Ignoring invalid contact point foobar (expecting host:port)");
+ assertLog(Level.WARN, "Ignoring invalid contact point foobar (expecting format host:port)");
}
@Test
@@ -132,7 +132,7 @@ public void should_ignore_malformed_port_and_warn() {
assertThat(endPoints).isEmpty();
assertLog(
Level.WARN,
- "Ignoring invalid contact point 127.0.0.1:foobar (expecting a number, got foobar)");
+ "Ignoring invalid contact point 127.0.0.1:foobar (expecting port to be a number, got foobar)");
}
@Test
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslatorTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslatorTest.java
index c5e864b4bae..3bb9c4bc291 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslatorTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslatorTest.java
@@ -17,6 +17,7 @@
*/
package com.datastax.oss.driver.internal.core.addresstranslation;
+import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -25,7 +26,6 @@
import com.datastax.oss.driver.internal.core.context.DefaultDriverContext;
import com.datastax.oss.driver.internal.core.context.MockedDriverContextFactory;
import java.net.InetSocketAddress;
-import java.util.Optional;
import org.junit.Test;
public class FixedHostNameAddressTranslatorTest {
@@ -33,11 +33,9 @@ public class FixedHostNameAddressTranslatorTest {
@Test
public void should_translate_address() {
DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
- when(defaultProfile.getString(
- FixedHostNameAddressTranslator.ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME_OPTION))
- .thenReturn("myaddress");
+ when(defaultProfile.getString(ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME)).thenReturn("myaddress");
DefaultDriverContext defaultDriverContext =
- MockedDriverContextFactory.defaultDriverContext(Optional.of(defaultProfile));
+ MockedDriverContextFactory.defaultDriverContext(defaultProfile);
FixedHostNameAddressTranslator translator =
new FixedHostNameAddressTranslator(defaultDriverContext);
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTest.java
new file mode 100644
index 00000000000..bd505f5dd44
--- /dev/null
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.addresstranslation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import java.net.InetSocketAddress;
+import org.junit.Test;
+
+public class SubnetAddressTest {
+ @Test
+ public void should_return_return_true_on_overlapping_with_another_subnet_address() {
+ SubnetAddress subnetAddress1 =
+ new SubnetAddress("100.64.0.0/15", mock(InetSocketAddress.class));
+ SubnetAddress subnetAddress2 =
+ new SubnetAddress("100.65.0.0/16", mock(InetSocketAddress.class));
+ assertThat(subnetAddress1.isOverlapping(subnetAddress2)).isTrue();
+ }
+
+ @Test
+ public void should_return_return_false_on_not_overlapping_with_another_subnet_address() {
+ SubnetAddress subnetAddress1 =
+ new SubnetAddress("100.64.0.0/15", mock(InetSocketAddress.class));
+ SubnetAddress subnetAddress2 =
+ new SubnetAddress("100.66.0.0/15", mock(InetSocketAddress.class));
+ assertThat(subnetAddress1.isOverlapping(subnetAddress2)).isFalse();
+ }
+}
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTranslatorTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTranslatorTest.java
new file mode 100644
index 00000000000..420170654dc
--- /dev/null
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTranslatorTest.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.addresstranslation;
+
+import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.ADDRESS_TRANSLATOR_DEFAULT_ADDRESS;
+import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.ADDRESS_TRANSLATOR_SUBNET_ADDRESSES;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
+import com.datastax.oss.driver.internal.core.context.DefaultDriverContext;
+import com.datastax.oss.driver.internal.core.context.MockedDriverContextFactory;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import org.junit.Test;
+
+@SuppressWarnings("resource")
+public class SubnetAddressTranslatorTest {
+
+ @Test
+ public void should_translate_to_correct_subnet_address_ipv4() {
+ Map subnetAddresses =
+ ImmutableMap.of(
+ "\"100.64.0.0/15\"", "cassandra.datacenter1.com:19042",
+ "100.66.0.\"0/15\"", "cassandra.datacenter2.com:19042");
+ DefaultDriverContext context = context(subnetAddresses);
+ SubnetAddressTranslator translator = new SubnetAddressTranslator(context);
+ InetSocketAddress address = new InetSocketAddress("100.64.0.1", 9042);
+ assertThat(translator.translate(address))
+ .isEqualTo(InetSocketAddress.createUnresolved("cassandra.datacenter1.com", 19042));
+ }
+
+ @Test
+ public void should_translate_to_correct_subnet_address_ipv6() {
+ Map subnetAddresses =
+ ImmutableMap.of(
+ "\"::ffff:6440:0/111\"", "cassandra.datacenter1.com:19042",
+ "\"::ffff:6442:0/111\"", "cassandra.datacenter2.com:19042");
+ DefaultDriverContext context = context(subnetAddresses);
+ SubnetAddressTranslator translator = new SubnetAddressTranslator(context);
+ InetSocketAddress address = new InetSocketAddress("::ffff:6440:1", 9042);
+ assertThat(translator.translate(address))
+ .isEqualTo(InetSocketAddress.createUnresolved("cassandra.datacenter1.com", 19042));
+ }
+
+ @Test
+ public void should_translate_to_default_address() {
+ DefaultDriverContext context = context(ImmutableMap.of());
+ when(context
+ .getConfig()
+ .getDefaultProfile()
+ .getString(ADDRESS_TRANSLATOR_DEFAULT_ADDRESS, null))
+ .thenReturn("cassandra.com:19042");
+ SubnetAddressTranslator translator = new SubnetAddressTranslator(context);
+ InetSocketAddress address = new InetSocketAddress("100.68.0.1", 9042);
+ assertThat(translator.translate(address))
+ .isEqualTo(InetSocketAddress.createUnresolved("cassandra.com", 19042));
+ }
+
+ @Test
+ public void should_pass_through_not_matched_address() {
+ DefaultDriverContext context = context(ImmutableMap.of());
+ SubnetAddressTranslator translator = new SubnetAddressTranslator(context);
+ InetSocketAddress address = new InetSocketAddress("100.68.0.1", 9042);
+ assertThat(translator.translate(address)).isEqualTo(address);
+ }
+
+ @Test
+ public void should_fail_on_intersecting_subnets_ipv4() {
+ Map subnetAddresses =
+ ImmutableMap.of(
+ "\"100.64.0.0/15\"", "cassandra.datacenter1.com:19042",
+ "100.65.0.\"0/16\"", "cassandra.datacenter2.com:19042");
+ DefaultDriverContext context = context(subnetAddresses);
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> new SubnetAddressTranslator(context))
+ .withMessage(
+ "Configured subnets are overlapping: "
+ + String.format(
+ "SubnetAddress[subnet=[100, 64, 0, 0], address=%s], ",
+ InetSocketAddress.createUnresolved("cassandra.datacenter1.com", 19042))
+ + String.format(
+ "SubnetAddress[subnet=[100, 65, 0, 0], address=%s]",
+ InetSocketAddress.createUnresolved("cassandra.datacenter2.com", 19042)));
+ }
+
+ @Test
+ public void should_fail_on_intersecting_subnets_ipv6() {
+ Map subnetAddresses =
+ ImmutableMap.of(
+ "\"::ffff:6440:0/111\"", "cassandra.datacenter1.com:19042",
+ "\"::ffff:6441:0/112\"", "cassandra.datacenter2.com:19042");
+ DefaultDriverContext context = context(subnetAddresses);
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> new SubnetAddressTranslator(context))
+ .withMessage(
+ "Configured subnets are overlapping: "
+ + String.format(
+ "SubnetAddress[subnet=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 100, 64, 0, 0], address=%s], ",
+ InetSocketAddress.createUnresolved("cassandra.datacenter1.com", 19042))
+ + String.format(
+ "SubnetAddress[subnet=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 100, 65, 0, 0], address=%s]",
+ InetSocketAddress.createUnresolved("cassandra.datacenter2.com", 19042)));
+ }
+
+ @Test
+ public void should_fail_on_subnet_address_without_port() {
+ Map subnetAddresses =
+ ImmutableMap.of("\"100.64.0.0/15\"", "cassandra.datacenter1.com");
+ DefaultDriverContext context = context(subnetAddresses);
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> new SubnetAddressTranslator(context))
+ .withMessage("Invalid address cassandra.datacenter1.com (expecting format host:port)");
+ }
+
+ @Test
+ public void should_fail_on_default_address_without_port() {
+ DefaultDriverContext context = context(ImmutableMap.of());
+ when(context
+ .getConfig()
+ .getDefaultProfile()
+ .getString(ADDRESS_TRANSLATOR_DEFAULT_ADDRESS, null))
+ .thenReturn("cassandra.com");
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> new SubnetAddressTranslator(context))
+ .withMessage("Invalid address cassandra.com (expecting format host:port)");
+ }
+
+ private static DefaultDriverContext context(Map subnetAddresses) {
+ DriverExecutionProfile profile = mock(DriverExecutionProfile.class);
+ when(profile.getStringMap(ADDRESS_TRANSLATOR_SUBNET_ADDRESSES)).thenReturn(subnetAddresses);
+ return MockedDriverContextFactory.defaultDriverContext(profile);
+ }
+}
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetTest.java
new file mode 100644
index 00000000000..f8ba8929e9e
--- /dev/null
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetTest.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.addresstranslation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+
+import java.net.UnknownHostException;
+import org.junit.Test;
+
+public class SubnetTest {
+ @Test
+ public void should_parse_to_correct_ipv4_subnet() throws UnknownHostException {
+ Subnet subnet = Subnet.parse("100.64.0.0/15");
+ assertThat(subnet.getSubnet()).containsExactly(100, 64, 0, 0);
+ assertThat(subnet.getNetworkMask()).containsExactly(255, 254, 0, 0);
+ assertThat(subnet.getUpper()).containsExactly(100, 65, 255, 255);
+ assertThat(subnet.getLower()).containsExactly(100, 64, 0, 0);
+ }
+
+ @Test
+ public void should_parse_to_correct_ipv6_subnet() throws UnknownHostException {
+ Subnet subnet = Subnet.parse("2001:db8:85a3::8a2e:370:0/111");
+ assertThat(subnet.getSubnet())
+ .containsExactly(32, 1, 13, 184, 133, 163, 0, 0, 0, 0, 138, 46, 3, 112, 0, 0);
+ assertThat(subnet.getNetworkMask())
+ .containsExactly(
+ 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 254, 0, 0);
+ assertThat(subnet.getUpper())
+ .containsExactly(32, 1, 13, 184, 133, 163, 0, 0, 0, 0, 138, 46, 3, 113, 255, 255);
+ assertThat(subnet.getLower())
+ .containsExactly(32, 1, 13, 184, 133, 163, 0, 0, 0, 0, 138, 46, 3, 112, 0, 0);
+ }
+
+ @Test
+ public void should_parse_to_correct_ipv6_subnet_ipv4_convertible() throws UnknownHostException {
+ Subnet subnet = Subnet.parse("::ffff:6440:0/111");
+ assertThat(subnet.getSubnet())
+ .containsExactly(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 100, 64, 0, 0);
+ assertThat(subnet.getNetworkMask())
+ .containsExactly(
+ 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 254, 0, 0);
+ assertThat(subnet.getUpper())
+ .containsExactly(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 100, 65, 255, 255);
+ assertThat(subnet.getLower())
+ .containsExactly(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 100, 64, 0, 0);
+ }
+
+ @Test
+ public void should_fail_on_invalid_cidr_format() {
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> Subnet.parse("invalid"))
+ .withMessage("Invalid subnet: invalid");
+ }
+
+ @Test
+ public void should_parse_bounding_prefix_lengths_correctly() {
+ assertThatNoException().isThrownBy(() -> Subnet.parse("0.0.0.0/0"));
+ assertThatNoException().isThrownBy(() -> Subnet.parse("100.64.0.0/32"));
+ }
+
+ @Test
+ public void should_fail_on_invalid_prefix_length() {
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> Subnet.parse("100.64.0.0/-1"))
+ .withMessage("Prefix length -1 must be within [0; 32]");
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> Subnet.parse("100.64.0.0/33"))
+ .withMessage("Prefix length 33 must be within [0; 32]");
+ }
+
+ @Test
+ public void should_fail_on_not_prefix_block_subnet_ipv4() {
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> Subnet.parse("100.65.0.0/15"))
+ .withMessage("Subnet 100.65.0.0/15 must be represented as a network prefix block");
+ }
+
+ @Test
+ public void should_fail_on_not_prefix_block_subnet_ipv6() {
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> Subnet.parse("::ffff:6441:0/111"))
+ .withMessage("Subnet ::ffff:6441:0/111 must be represented as a network prefix block");
+ }
+
+ @Test
+ public void should_return_true_on_containing_address() throws UnknownHostException {
+ Subnet subnet = Subnet.parse("100.64.0.0/15");
+ assertThat(subnet.contains(new byte[] {100, 64, 0, 0})).isTrue();
+ assertThat(subnet.contains(new byte[] {100, 65, (byte) 255, (byte) 255})).isTrue();
+ assertThat(subnet.contains(new byte[] {100, 65, 100, 100})).isTrue();
+ }
+
+ @Test
+ public void should_return_false_on_not_containing_address() throws UnknownHostException {
+ Subnet subnet = Subnet.parse("100.64.0.0/15");
+ assertThat(subnet.contains(new byte[] {100, 63, (byte) 255, (byte) 255})).isFalse();
+ assertThat(subnet.contains(new byte[] {100, 66, 0, 0})).isFalse();
+ // IPv6 cannot be contained by IPv4 subnet.
+ assertThat(subnet.contains(new byte[16])).isFalse();
+ }
+}
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/config/MockOptions.java b/core/src/test/java/com/datastax/oss/driver/internal/core/config/MockOptions.java
index 25c1e8b26fd..cee57abbfdf 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/config/MockOptions.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/config/MockOptions.java
@@ -24,6 +24,7 @@ public enum MockOptions implements DriverOption {
INT1("int1"),
INT2("int2"),
AUTH_PROVIDER("auth_provider"),
+ SUBNET_ADDRESSES("subnet_addresses"),
;
private final String path;
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/config/typesafe/TypesafeDriverConfigTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/config/typesafe/TypesafeDriverConfigTest.java
index 16ccb73da9f..4a78c3ccb03 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/config/typesafe/TypesafeDriverConfigTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/config/typesafe/TypesafeDriverConfigTest.java
@@ -101,7 +101,6 @@ public void should_fetch_string_map() {
parse(
"int1 = 42 \n auth_provider { auth_thing_one= one \n auth_thing_two = two \n auth_thing_three = three}");
DriverExecutionProfile base = config.getDefaultProfile();
- base.getStringMap(MockOptions.AUTH_PROVIDER);
Map map = base.getStringMap(MockOptions.AUTH_PROVIDER);
assertThat(map.entrySet().size()).isEqualTo(3);
assertThat(map.get("auth_thing_one")).isEqualTo("one");
@@ -109,6 +108,19 @@ public void should_fetch_string_map() {
assertThat(map.get("auth_thing_three")).isEqualTo("three");
}
+ @Test
+ public void should_fetch_string_map_with_forward_slash_in_keys() {
+ TypesafeDriverConfig config =
+ parse(
+ "subnet_addresses { 100.64.0.0/15 = \"cassandra.datacenter1.com:9042\" \n \"100.66.0.0/15\" = \"cassandra.datacenter2.com\" \n \"::ffff:6440:0/111\" = \"cassandra.datacenter3.com:19042\" }");
+ DriverExecutionProfile base = config.getDefaultProfile();
+ Map map = base.getStringMap(MockOptions.SUBNET_ADDRESSES);
+ assertThat(map.entrySet().size()).isEqualTo(3);
+ assertThat(map.get("100.64.0.\"0/15\"")).isEqualTo("cassandra.datacenter1.com:9042");
+ assertThat(map.get("\"100.66.0.0/15\"")).isEqualTo("cassandra.datacenter2.com");
+ assertThat(map.get("\"::ffff:6440:0/111\"")).isEqualTo("cassandra.datacenter3.com:19042");
+ }
+
@Test
public void should_create_derived_profile_with_string_map() {
TypesafeDriverConfig config = parse("int1 = 42");
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContextTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContextTest.java
index baf101508d4..6d4585cb4d7 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContextTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContextTest.java
@@ -42,7 +42,7 @@ private DefaultDriverContext buildMockedContext(Optional compressionOpti
DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
when(defaultProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
.thenReturn(compressionOption.orElse("none"));
- return MockedDriverContextFactory.defaultDriverContext(Optional.of(defaultProfile));
+ return MockedDriverContextFactory.defaultDriverContext(defaultProfile);
}
private void doCreateCompressorTest(Optional configVal, Class expectedClz) {
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/context/MockedDriverContextFactory.java b/core/src/test/java/com/datastax/oss/driver/internal/core/context/MockedDriverContextFactory.java
index 06817326844..a27d3449519 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/context/MockedDriverContextFactory.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/context/MockedDriverContextFactory.java
@@ -24,44 +24,45 @@
import com.datastax.oss.driver.api.core.config.DriverConfig;
import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
+import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy;
+import com.datastax.oss.driver.api.core.loadbalancing.NodeDistanceEvaluator;
+import com.datastax.oss.driver.api.core.metadata.Node;
import com.datastax.oss.driver.api.core.metadata.NodeStateListener;
import com.datastax.oss.driver.api.core.metadata.schema.SchemaChangeListener;
import com.datastax.oss.driver.api.core.session.ProgrammaticArguments;
import com.datastax.oss.driver.api.core.tracker.RequestTracker;
+import com.datastax.oss.driver.internal.core.ConsistencyLevelRegistry;
+import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
import com.datastax.oss.driver.shaded.guava.common.collect.Maps;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.UUID;
public class MockedDriverContextFactory {
public static DefaultDriverContext defaultDriverContext() {
- return defaultDriverContext(Optional.empty());
+ return defaultDriverContext(MockedDriverContextFactory.defaultProfile("datacenter1"));
}
public static DefaultDriverContext defaultDriverContext(
- Optional profileOption) {
-
- /* If the caller provided a profile use that, otherwise make a new one */
- final DriverExecutionProfile profile =
- profileOption.orElseGet(
- () -> {
- DriverExecutionProfile blankProfile = mock(DriverExecutionProfile.class);
- when(blankProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
- .thenReturn("none");
- when(blankProfile.getDuration(DefaultDriverOption.METRICS_NODE_EXPIRE_AFTER))
- .thenReturn(Duration.ofMinutes(5));
- when(blankProfile.isDefined(DefaultDriverOption.METRICS_FACTORY_CLASS))
- .thenReturn(true);
- when(blankProfile.getString(DefaultDriverOption.METRICS_FACTORY_CLASS))
- .thenReturn("DefaultMetricsFactory");
- return blankProfile;
- });
+ DriverExecutionProfile defaultProfile, DriverExecutionProfile... profiles) {
/* Setup machinery to connect the input DriverExecutionProfile to the config loader */
final DriverConfig driverConfig = mock(DriverConfig.class);
final DriverConfigLoader configLoader = mock(DriverConfigLoader.class);
when(configLoader.getInitialConfig()).thenReturn(driverConfig);
- when(driverConfig.getDefaultProfile()).thenReturn(profile);
+ when(driverConfig.getDefaultProfile()).thenReturn(defaultProfile);
+ when(driverConfig.getProfile(defaultProfile.getName())).thenReturn(defaultProfile);
+
+ for (DriverExecutionProfile profile : profiles) {
+ when(driverConfig.getProfile(profile.getName())).thenReturn(profile);
+ }
ProgrammaticArguments args =
ProgrammaticArguments.builder()
@@ -71,6 +72,92 @@ public static DefaultDriverContext defaultDriverContext(
.withLocalDatacenters(Maps.newHashMap())
.withNodeDistanceEvaluators(Maps.newHashMap())
.build();
- return new DefaultDriverContext(configLoader, args);
+
+ return new DefaultDriverContext(configLoader, args) {
+ @NonNull
+ @Override
+ public Map getLoadBalancingPolicies() {
+ ImmutableMap.Builder map = ImmutableMap.builder();
+ map.put(
+ defaultProfile.getName(),
+ mockLoadBalancingPolicy(
+ this,
+ defaultProfile.getName(),
+ defaultProfile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)));
+ for (DriverExecutionProfile profile : profiles) {
+ map.put(
+ profile.getName(),
+ mockLoadBalancingPolicy(
+ this,
+ profile.getName(),
+ profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)));
+ }
+ return map.build();
+ }
+
+ @NonNull
+ @Override
+ public ConsistencyLevelRegistry getConsistencyLevelRegistry() {
+ return mock(ConsistencyLevelRegistry.class);
+ }
+ };
+ }
+
+ public static DriverExecutionProfile defaultProfile(String localDc) {
+ return createProfile(DriverExecutionProfile.DEFAULT_NAME, localDc);
+ }
+
+ public static DriverExecutionProfile createProfile(String name, String localDc) {
+ DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
+ when(defaultProfile.getName()).thenReturn(name);
+ when(defaultProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
+ .thenReturn("none");
+ when(defaultProfile.getDuration(DefaultDriverOption.METRICS_NODE_EXPIRE_AFTER))
+ .thenReturn(Duration.ofMinutes(5));
+ when(defaultProfile.isDefined(DefaultDriverOption.METRICS_FACTORY_CLASS)).thenReturn(true);
+ when(defaultProfile.getString(DefaultDriverOption.METRICS_FACTORY_CLASS))
+ .thenReturn("DefaultMetricsFactory");
+ when(defaultProfile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER))
+ .thenReturn(localDc);
+ when(defaultProfile.getString(
+ DefaultDriverOption.LOAD_BALANCING_DEFAULT_LWT_REQUEST_ROUTING_METHOD))
+ .thenReturn("PRESERVE_REPLICA_ORDER");
+ return defaultProfile;
+ }
+
+ public static void allowRemoteDcConnectivity(
+ DriverExecutionProfile profile,
+ int maxNodesPerRemoteDc,
+ boolean allowRemoteSatisfyLocalDc,
+ List preferredRemoteDcs) {
+ when(profile.getInt(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_MAX_NODES_PER_REMOTE_DC))
+ .thenReturn(maxNodesPerRemoteDc);
+ when(profile.getBoolean(
+ DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_ALLOW_FOR_LOCAL_CONSISTENCY_LEVELS))
+ .thenReturn(allowRemoteSatisfyLocalDc);
+ when(profile.getStringList(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_PREFERRED_REMOTE_DCS))
+ .thenReturn(preferredRemoteDcs);
+ }
+
+ private static LoadBalancingPolicy mockLoadBalancingPolicy(
+ DefaultDriverContext driverContext, String profile, String localDc) {
+ LoadBalancingPolicy loadBalancingPolicy =
+ new DefaultLoadBalancingPolicy(driverContext, profile) {
+ @NonNull
+ @Override
+ protected Optional discoverLocalDc(@NonNull Map nodes) {
+ return Optional.ofNullable(localDc);
+ }
+
+ @NonNull
+ @Override
+ protected NodeDistanceEvaluator createNodeDistanceEvaluator(
+ @Nullable String localDc, @NonNull Map nodes) {
+ return mock(NodeDistanceEvaluator.class);
+ }
+ };
+ loadBalancingPolicy.init(
+ Collections.emptyMap(), mock(LoadBalancingPolicy.DistanceReporter.class));
+ return loadBalancingPolicy;
}
}
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilderTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilderTest.java
index 33811b2793a..d12e50b7e8e 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilderTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilderTest.java
@@ -26,10 +26,10 @@
import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
import com.datastax.oss.driver.api.core.session.Session;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
import com.datastax.oss.protocol.internal.request.Startup;
import com.tngtech.java.junit.dataprovider.DataProvider;
import com.tngtech.java.junit.dataprovider.DataProviderRunner;
-import java.util.Optional;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -41,7 +41,8 @@ private DefaultDriverContext buildMockedContext(String compression) {
DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
when(defaultProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
.thenReturn(compression);
- return MockedDriverContextFactory.defaultDriverContext(Optional.of(defaultProfile));
+ when(defaultProfile.getName()).thenReturn(DriverExecutionProfile.DEFAULT_NAME);
+ return MockedDriverContextFactory.defaultDriverContext(defaultProfile);
}
private void assertDefaultStartupOptions(Startup startup) {
@@ -94,4 +95,44 @@ public void should_fail_to_build_startup_options_with_invalid_compression() {
new Startup(ctx.getStartupOptions());
});
}
+
+ @Test
+ public void should_include_all_local_dcs_in_startup_message() {
+
+ DefaultDriverContext ctx =
+ MockedDriverContextFactory.defaultDriverContext(
+ MockedDriverContextFactory.defaultProfile("us-west-2"),
+ MockedDriverContextFactory.createProfile("oltp", "us-east-2"),
+ MockedDriverContextFactory.createProfile("olap", "eu-central-1"));
+ Startup startup = new Startup(ctx.getStartupOptions());
+ assertThat(startup.options)
+ .containsEntry(
+ StartupOptionsBuilder.DRIVER_BAGGAGE,
+ "{\"default\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"us-west-2\"}},"
+ + "\"oltp\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"us-east-2\"}},"
+ + "\"olap\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"eu-central-1\"}}}");
+ }
+
+ @Test
+ public void should_include_all_lbp_details_in_startup_message() {
+
+ DriverExecutionProfile defaultProfile = MockedDriverContextFactory.defaultProfile("dc1");
+ DriverExecutionProfile oltpProfile = MockedDriverContextFactory.createProfile("oltp", "dc1");
+ MockedDriverContextFactory.allowRemoteDcConnectivity(
+ oltpProfile, 2, true, ImmutableList.of("dc2", "dc3"));
+ DefaultDriverContext ctx =
+ MockedDriverContextFactory.defaultDriverContext(defaultProfile, oltpProfile);
+
+ Startup startup = new Startup(ctx.getStartupOptions());
+
+ assertThat(startup.options)
+ .containsEntry(
+ StartupOptionsBuilder.DRIVER_BAGGAGE,
+ "{\"default\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"dc1\"}},"
+ + "\"oltp\":{\"DefaultLoadBalancingPolicy\":{"
+ + "\"localDc\":\"dc1\","
+ + "\"preferredRemoteDcs\":[\"dc2\",\"dc3\"],"
+ + "\"allowDcFailoverForLocalCl\":true,"
+ + "\"maxNodesPerRemoteDc\":2}}}");
+ }
}
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerRetryTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerRetryTest.java
index bea52891c18..ccac873c616 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerRetryTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerRetryTest.java
@@ -48,6 +48,8 @@
import com.datastax.oss.driver.api.core.servererrors.ServerError;
import com.datastax.oss.driver.api.core.servererrors.UnavailableException;
import com.datastax.oss.driver.api.core.servererrors.WriteTimeoutException;
+import com.datastax.oss.driver.api.core.session.Request;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
import com.datastax.oss.protocol.internal.ProtocolConstants;
import com.datastax.oss.protocol.internal.response.Error;
import com.datastax.oss.protocol.internal.response.error.ReadTimeout;
@@ -55,9 +57,13 @@
import com.datastax.oss.protocol.internal.response.error.WriteTimeout;
import com.tngtech.java.junit.dataprovider.DataProvider;
import com.tngtech.java.junit.dataprovider.UseDataProvider;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
public class CqlRequestHandlerRetryTest extends CqlRequestHandlerTestBase {
@@ -384,6 +390,63 @@ public void should_rethrow_error_if_not_idempotent_and_error_unsafe_or_policy_re
}
}
+ @Test
+ @UseDataProvider("failureAndIdempotent")
+ public void should_not_fail_with_duplicate_key_when_retrying_with_request_id_generator(
+ FailureScenario failureScenario, boolean defaultIdempotence, Statement statement) {
+
+ // Create a RequestIdGenerator that uses the same key as the statement's custom payload
+ RequestIdGenerator requestIdGenerator =
+ new RequestIdGenerator() {
+ private AtomicInteger counter = new AtomicInteger(0);
+
+ @Override
+ public String getSessionRequestId() {
+ return "session-123";
+ }
+
+ @Override
+ public String getNodeRequestId(@NonNull Request request, @NonNull String parentId) {
+ return parentId + "-" + counter.getAndIncrement();
+ }
+ };
+
+ RequestHandlerTestHarness.Builder harnessBuilder =
+ RequestHandlerTestHarness.builder()
+ .withDefaultIdempotence(defaultIdempotence)
+ .withRequestIdGenerator(requestIdGenerator);
+ failureScenario.mockRequestError(harnessBuilder, node1);
+ harnessBuilder.withResponse(node2, defaultFrameOf(singleRow()));
+
+ try (RequestHandlerTestHarness harness = harnessBuilder.build()) {
+ failureScenario.mockRetryPolicyVerdict(
+ harness.getContext().getRetryPolicy(anyString()), RetryVerdict.RETRY_NEXT);
+
+ CompletionStage resultSetFuture =
+ new CqlRequestHandler(statement, harness.getSession(), harness.getContext(), "test")
+ .handle();
+
+ // The test should succeed without throwing a duplicate key exception
+ assertThatStage(resultSetFuture)
+ .isSuccess(
+ resultSet -> {
+ Iterator rows = resultSet.currentPage().iterator();
+ assertThat(rows.hasNext()).isTrue();
+ assertThat(rows.next().getString("message")).isEqualTo("hello, world");
+
+ ExecutionInfo executionInfo = resultSet.getExecutionInfo();
+ assertThat(executionInfo.getCoordinator()).isEqualTo(node2);
+ assertThat(executionInfo.getErrors()).hasSize(1);
+ assertThat(executionInfo.getErrors().get(0).getKey()).isEqualTo(node1);
+
+ // Verify that the custom payload still contains the request ID key
+ // (either the original value or the generated one, depending on implementation)
+ assertThat(executionInfo.getRequest().getCustomPayload().get("request-id"))
+ .isEqualTo(ByteBuffer.wrap("session-123-1".getBytes(StandardCharsets.UTF_8)));
+ });
+ }
+ }
+
/**
* Sets up the mocks to simulate an error from a node, and make the retry policy return a given
* decision for that error.
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/QueryTraceFetcherTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/QueryTraceFetcherTest.java
index b355e0fc9f0..dc238775bc1 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/QueryTraceFetcherTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/QueryTraceFetcherTest.java
@@ -79,7 +79,7 @@ public class QueryTraceFetcherTest {
@Mock private NettyOptions nettyOptions;
@Mock private EventExecutorGroup adminEventExecutorGroup;
@Mock private EventExecutor eventExecutor;
- @Mock private InetAddress address;
+ private InetAddress address = InetAddress.getLoopbackAddress();
@Captor private ArgumentCaptor statementCaptor;
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java
index 3ec30286755..8ddc66c9c4e 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java
@@ -44,6 +44,7 @@
import com.datastax.oss.driver.api.core.session.Session;
import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy;
import com.datastax.oss.driver.api.core.time.TimestampGenerator;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
import com.datastax.oss.driver.internal.core.DefaultConsistencyLevelRegistry;
import com.datastax.oss.driver.internal.core.ProtocolFeature;
@@ -69,6 +70,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -194,6 +196,9 @@ protected RequestHandlerTestHarness(Builder builder) {
when(context.getRequestThrottler()).thenReturn(new PassThroughRequestThrottler(context));
when(context.getRequestTracker()).thenReturn(new NoopRequestTracker(context));
+
+ when(context.getRequestIdGenerator())
+ .thenReturn(Optional.ofNullable(builder.requestIdGenerator));
}
public DefaultSession getSession() {
@@ -226,6 +231,7 @@ public static class Builder {
private final List poolBehaviors = new ArrayList<>();
private boolean defaultIdempotence;
private ProtocolVersion protocolVersion;
+ private RequestIdGenerator requestIdGenerator;
/**
* Sets the given node as the next one in the query plan; an empty pool will be simulated when
@@ -281,6 +287,11 @@ public Builder withProtocolVersion(ProtocolVersion protocolVersion) {
return this;
}
+ public Builder withRequestIdGenerator(RequestIdGenerator requestIdGenerator) {
+ this.requestIdGenerator = requestIdGenerator;
+ return this;
+ }
+
/**
* Sets the given node as the next one in the query plan; the test code is responsible of
* calling the methods on the returned object to complete the write and the query.
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/tracker/RequestIdGeneratorTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/tracker/RequestIdGeneratorTest.java
new file mode 100644
index 00000000000..fb1883e125f
--- /dev/null
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/tracker/RequestIdGeneratorTest.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.tracker;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.datastax.oss.driver.api.core.cql.Statement;
+import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
+import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.Strict.class)
+public class RequestIdGeneratorTest {
+ @Mock private InternalDriverContext context;
+ @Mock private Statement statement;
+
+ @Test
+ public void uuid_generator_should_generate() {
+ // given
+ UuidRequestIdGenerator generator = new UuidRequestIdGenerator(context);
+ // when
+ String parentId = generator.getSessionRequestId();
+ String requestId = generator.getNodeRequestId(statement, parentId);
+ // then
+ // e.g. "550e8400-e29b-41d4-a716-446655440000", which is 36 characters long
+ assertThat(parentId.length()).isEqualTo(36);
+ // e.g. "550e8400-e29b-41d4-a716-446655440000-550e8400-e29b-41d4-a716-446655440000", which is 73
+ // characters long
+ assertThat(requestId.length()).isEqualTo(73);
+ }
+
+ @Test
+ public void w3c_generator_should_generate() {
+ // given
+ W3CContextRequestIdGenerator generator = new W3CContextRequestIdGenerator(context);
+ // when
+ String parentId = generator.getSessionRequestId();
+ String requestId = generator.getNodeRequestId(statement, parentId);
+ // then
+ // e.g. "4bf92f3577b34da6a3ce929d0e0e4736", which is 32 characters long
+ assertThat(parentId.length()).isEqualTo(32);
+ // According to W3C "traceparent" spec,
+ // https://www.w3.org/TR/trace-context/#traceparent-header-field-values
+ // e.g. "00-4bf92f3577b34da6a3ce929d0e0e4736-a3ce929d0e0e4736-01", which 55 characters long
+ assertThat(requestId.length()).isEqualTo(55);
+ }
+
+ @Test
+ public void w3c_generator_default_payloadkey() {
+ W3CContextRequestIdGenerator w3cGenerator = new W3CContextRequestIdGenerator(context);
+ assertThat(w3cGenerator.getCustomPayloadKey())
+ .isEqualTo(RequestIdGenerator.DEFAULT_PAYLOAD_KEY);
+ }
+
+ @Test
+ public void w3c_generator_provided_payloadkey() {
+ String someString = RandomStringUtils.random(12);
+ W3CContextRequestIdGenerator w3cGenerator = new W3CContextRequestIdGenerator(someString);
+ assertThat(w3cGenerator.getCustomPayloadKey()).isEqualTo(someString);
+ }
+}
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/VectorCodecTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/VectorCodecTest.java
index fe8e9e7be0f..17c78514127 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/VectorCodecTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/VectorCodecTest.java
@@ -34,8 +34,8 @@
import com.tngtech.java.junit.dataprovider.UseDataProvider;
import java.nio.ByteBuffer;
import java.time.LocalTime;
-import java.util.Arrays;
import java.util.HashMap;
+import org.apache.commons.lang3.ArrayUtils;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -128,10 +128,7 @@ public void should_throw_on_encode_with_too_few_elements(TestDataContainer testD
@Test
@UseDataProvider("dataProvider")
public void should_throw_on_encode_with_too_many_elements(TestDataContainer testData) {
- Object[] first = testData.getValues();
- Object[] second = testData.getValues();
- Object[] doubled = Arrays.copyOf(first, first.length + second.length);
- System.arraycopy(second, 0, doubled, first.length, second.length);
+ Object[] doubled = ArrayUtils.addAll(testData.getValues(), testData.getValues());
TypeCodec> codec = getCodec(testData.getDataType());
assertThatThrownBy(() -> codec.encode(CqlVector.newInstance(doubled), ProtocolVersion.DEFAULT))
.isInstanceOf(IllegalArgumentException.class);
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/util/ArrayUtilsTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/util/ArrayUtilsTest.java
index c2a7fb70304..c2df6449fdb 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/util/ArrayUtilsTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/util/ArrayUtilsTest.java
@@ -22,7 +22,7 @@
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-import java.util.concurrent.ThreadLocalRandom;
+import java.util.Random;
import org.junit.Test;
public class ArrayUtilsTest {
@@ -86,7 +86,7 @@ public void should_not_bubble_down_when_target_index_lower() {
@Test
public void should_shuffle_head() {
String[] array = {"a", "b", "c", "d", "e"};
- ThreadLocalRandom random = mock(ThreadLocalRandom.class);
+ Random random = mock(Random.class);
when(random.nextInt(anyInt()))
.thenAnswer(
(invocation) -> {
diff --git a/distribution-source/pom.xml b/distribution-source/pom.xml
index 19ad9f68bf1..23256425f31 100644
--- a/distribution-source/pom.xml
+++ b/distribution-source/pom.xml
@@ -28,7 +28,7 @@
com.scylladb
java-driver-parent
- 4.19.0.10-SNAPSHOT
+ 4.19.2.0-SNAPSHOT
java-driver-distribution-source
pom
diff --git a/distribution-tests/pom.xml b/distribution-tests/pom.xml
index 5ff2f590dc0..4996f658ab7 100644
--- a/distribution-tests/pom.xml
+++ b/distribution-tests/pom.xml
@@ -28,7 +28,7 @@
com.scylladb
java-driver-parent
- 4.19.0.10-SNAPSHOT
+ 4.19.2.0-SNAPSHOT
java-driver-distribution-tests
Scylla Java Driver - distribution tests
diff --git a/distribution/pom.xml b/distribution/pom.xml
index cb38a8c8475..bf86d2e1fe8 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -28,7 +28,7 @@
com.scylladb
java-driver-parent
- 4.19.0.10-SNAPSHOT
+ 4.19.2.0-SNAPSHOT
java-driver-distribution
diff --git a/examples/pom.xml b/examples/pom.xml
index 19e70197168..0d9a5c64373 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -28,7 +28,7 @@
java-driver-parent
com.scylladb
- 4.19.0.10-SNAPSHOT
+ 4.19.2.0-SNAPSHOT
java-driver-examples
Java driver for Scylla and Apache Cassandra(R) - examples.
diff --git a/guava-shaded/pom.xml b/guava-shaded/pom.xml
index 277024a2dd5..27d5b69077d 100644
--- a/guava-shaded/pom.xml
+++ b/guava-shaded/pom.xml
@@ -23,7 +23,7 @@
com.scylladb
java-driver-parent
- 4.19.0.10-SNAPSHOT
+ 4.19.2.0-SNAPSHOT
java-driver-guava-shaded
Scylla Java Driver - guava shaded dep
@@ -46,6 +46,7 @@
error_prone_annotations
+ true