diff --git a/core/pom.xml b/core/pom.xml index 8342b8b6df5..0efeb510b1c 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -160,6 +160,18 @@ mockito-core test + + com.networknt + json-schema-validator + test + + + + me.fabriciorby + maven-surefire-junit5-tree-reporter + + + io.reactivex.rxjava2 rxjava 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 3a6e4ed69bb..970ab306d07 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 @@ -1175,16 +1175,21 @@ public enum DefaultDriverOption implements DriverOption { */ ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses"), /** - * Whether the driver reports its effective configuration to ScyllaDB at connection time. - * - *

When {@code true}, the driver adds two entries to the CQL {@code STARTUP} options, which - * ScyllaDB stores in {@code system.clients.client_options} so operators can inspect driver - * settings while investigating incidents: a {@code SESSION_ID} on every connection (so the server - * can group a session's connections) and a compact JSON payload under the {@code DRIVER_CONFIG} - * key on the control connection only. At this stage the {@code DRIVER_CONFIG} payload carries - * only schema-version metadata ({"version":1}); reporting of the effective - * configuration fields is planned for a later stage. When {@code false}, neither entry is sent - * and there is no change on the wire. + * Whether the driver reports its effective configuration to the cluster at connection time. + * Defaults to {@code true}. + * + *

When {@code true}, the control connection adds a compact JSON payload under the {@code + * DRIVER_CONFIG} key to its CQL {@code STARTUP} options, which the server stores in its + * client-connection system table ({@code system.clients} on ScyllaDB, {@code + * system_views.clients} on Cassandra 4.1+) so operators can inspect driver settings while + * investigating incidents. It describes the effective configuration of the driver's default + * execution profile (connection/socket settings, timeouts, retry/reconnection/ + * speculative-execution/load-balancing policies, connection pooling, query defaults, and TLS). + * Only the control connection sends it, since it describes the whole session. When {@code false}, + * {@code DRIVER_CONFIG} is not sent. + * + *

This option does not govern the {@code SESSION_ID} startup option, which the driver always + * sends on every connection so that the server can group a session's connections. * *

Value type: boolean */ diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java index 1d906caa985..c1a428b3524 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java @@ -400,7 +400,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) { // values) with no sensible scalar default, analogous to how CONFIG_RELOAD_INTERVAL is omitted. map.put(TypedDriverOption.CLIENT_ROUTES_NATIVE_TRANSPORT_PORT, 9042); map.put(TypedDriverOption.CLIENT_ROUTES_SHARD_AWARENESS_ENABLED, false); - map.put(TypedDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, false); + map.put(TypedDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, true); } @Immutable 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 e412b99b404..af93e734ef1 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 @@ -976,7 +976,7 @@ public String toString() { new TypedDriverOption<>( DefaultDriverOption.CLIENT_ROUTES_SHARD_AWARENESS_ENABLED, GenericType.BOOLEAN); - /** Whether the driver reports its configuration to ScyllaDB at connection time. */ + /** Whether the driver reports its configuration to the cluster at connection time. */ public static final TypedDriverOption DRIVER_CONFIG_REPORTING_ENABLED = new TypedDriverOption<>( DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, GenericType.BOOLEAN); 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 d65eaa864aa..806d84144ab 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 @@ -133,6 +133,11 @@ public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) { return engine; } + @Override + public boolean requireHostnameValidation() { + return requireHostnameValidation; + } + @Override public void close() { // nothing to do diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/ssl/SslEngineFactory.java b/core/src/main/java/com/datastax/oss/driver/api/core/ssl/SslEngineFactory.java index db4f18a97b9..2ec4144caa4 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/ssl/SslEngineFactory.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/ssl/SslEngineFactory.java @@ -37,4 +37,23 @@ public interface SslEngineFactory extends AutoCloseable { */ @NonNull SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint); + + /** + * Whether this factory validates the server certificate against the node's host name. + * + *

This is a diagnostic accessor (reported in the driver-configuration blob sent to the server + * at connection time); it does not affect how {@link #newSslEngine} behaves. + * + *

This method's default implementation returns {@code false}. The only reason it exists is to + * preserve binary compatibility. The driver's built-in factories override it to return their real + * value; the default is intentionally conservative because the driver cannot assume an arbitrary + * custom factory performs host name validation, and must not over-report a security control that + * may not actually be active. Custom factories that do validate should override this to report + * accurately. + * + * @since 4.19.2.1 + */ + default boolean requireHostnameValidation() { + return false; + } } diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java index cf782096f18..f2c6b95c54d 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java @@ -191,14 +191,26 @@ Message getRequest() { return request = Options.INSTANCE; case STARTUP: Map startupOptions = new HashMap<>(context.getStartupOptions()); + // Non-null sharding info is the driver's own proxy check for "this is ScyllaDB" (also + // used, independently, by CassandraSchemaQueries.shouldApplyUsingTimeout()). Detection + // only works once the OPTIONS/SUPPORTED handshake has populated featureStore, i.e. when + // querySupportedOptions is true (always the case today: ChannelFactory passes true for + // every connection) — featureStore itself is never actually null here. + boolean scyllaDb = false; if (featureStore != null) { featureStore.populateStartupOptions(startupOptions); + scyllaDb = featureStore.getShardingInfo() != null; + } + // The DRIVER_CONFIG blob describes the whole session, so only the control connection + // carries it (options.reportConfig); the other connections are correlated to it by the + // SESSION_ID that every connection already carries from context.getStartupOptions(). + // scyllaDb lets the report reflect ScyllaDB-only server-side behavior, e.g. the USING + // TIMEOUT clause on schema queries. No-op when driver config reporting is disabled. + if (options.reportConfig) { + context + .getDriverConfigReporter() + .populateControlConnectionOptions(startupOptions, scyllaDb); } - // Adds SESSION_ID on every connection and DRIVER_CONFIG on the control connection - // (options.reportConfig); no-op when driver config reporting is disabled. - context - .getDriverConfigReporter() - .populateStartupOptions(startupOptions, options.reportConfig); return request = new Startup(startupOptions); case GET_CLUSTER_NAME: return request = CLUSTER_NAME_QUERY; diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java index 2cd1b5a8560..295a4242bff 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java @@ -17,14 +17,30 @@ */ package com.datastax.oss.driver.internal.core.context; +import com.datastax.dse.driver.internal.core.loadbalancing.DseDcInferringLoadBalancingPolicy; +import com.datastax.dse.driver.internal.core.loadbalancing.DseLoadBalancingPolicy; 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.uuid.Uuids; +import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy; +import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy; +import com.datastax.oss.driver.api.core.retry.RetryPolicy; +import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy; +import com.datastax.oss.driver.api.core.ssl.SslEngineFactory; +import com.datastax.oss.driver.internal.core.connection.ConstantReconnectionPolicy; +import com.datastax.oss.driver.internal.core.connection.ExponentialReconnectionPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.BasicLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DcInferringLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.retry.ConsistencyDowngradingRetryPolicy; +import com.datastax.oss.driver.internal.core.retry.DefaultRetryPolicy; +import com.datastax.oss.driver.internal.core.specex.ConstantSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.specex.NoSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.time.ServerSideTimestampGenerator; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import java.util.Map; -import java.util.UUID; +import java.util.Optional; import net.jcip.annotations.ThreadSafe; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,8 +49,33 @@ * Default {@link DriverConfigReporter}: serializes the driver configuration to the cross-driver * {@code DRIVER_CONFIG} JSON shape and adds it to the control connection's {@code STARTUP} options. * - *

The blob is (re)built on demand every time the control connection initializes, so it always - * reflects the current (possibly reloaded) configuration without any caching. + *

Only the configuration blob is this class's concern; the {@link + * StartupOptionsBuilder#SESSION_ID_KEY SESSION_ID} that ties a session's connections together is an + * innate startup option, built elsewhere and sent regardless of whether reporting is enabled. + * + *

The blob is (re)built on demand every time the control connection initializes, so every group + * here always reads the current (possibly reloaded) {@link DriverExecutionProfile} at report time — + * never a cached field off a policy object. The running policy instances themselves may + * not be so current, though: some (e.g. {@code ExponentialReconnectionPolicy}'s backoff delays, or + * {@code DefaultLoadBalancingPolicy}'s slow-avoidance flag behind {@code latency-awareness}) cache + * the config value they were constructed with and don't re-read it on a live reload. So immediately + * after a reload, the report can show a value the already-running policy doesn't reflect yet — + * until that policy is rebuilt (e.g. a new instance takes over on the next reconnect). + * + *

Follows the schema's omission principle throughout: a key the Java driver has no equivalent + * for is left out of the JSON entirely rather than reported as {@code null}. + * + *

Known limitation: the report always describes {@link + * com.datastax.oss.driver.api.core.config.DriverExecutionProfile#DEFAULT_NAME the default execution + * profile}, not whichever profile a given request actually runs with. A session that relies on + * named execution profiles for some of its traffic will have that traffic's real settings + * (consistency level, timeouts, retry policy, ...) differ from what {@code DRIVER_CONFIG} reports. + * Reporting per-profile configuration would need a schema shape for multiple profiles, which the + * cross-driver schema doesn't define; this is a known gap, not an oversight. + * + *

Thread safety: this class is safe to use as shipped, and holds no mutable state. Note + * that {@code buildJson()} runs on every control-connection (re)initialization, and may be called + * concurrently with a reconnect racing a fresh session start. */ @ThreadSafe public class DefaultDriverConfigReporter implements DriverConfigReporter { @@ -44,9 +85,6 @@ public class DefaultDriverConfigReporter implements DriverConfigReporter { /** STARTUP option key under which the config JSON is sent. */ public static final String DRIVER_CONFIG_KEY = "DRIVER_CONFIG"; - /** STARTUP option key under which the per-session identifier is sent. */ - public static final String SESSION_ID_KEY = "SESSION_ID"; - /** * Major schema version. Adding keys is backward-compatible and does not bump this; only * changing/removing the meaning of an existing key does. @@ -57,60 +95,62 @@ public class DefaultDriverConfigReporter implements DriverConfigReporter { protected final InternalDriverContext context; - // Dedicated, driver-generated identifier for this session. Not derived from the (user-settable, - // Insights-oriented) CLIENT_ID, so that it is guaranteed unique per session as the grouping key - // requires. The reporter is a per-session singleton (built once via LazyReference), so this value - // is stable and shared across all of the session's connections. - private final UUID sessionId = Uuids.random(); - public DefaultDriverConfigReporter(InternalDriverContext context) { this.context = context; } @Override - public void populateStartupOptions( - Map startupOptions, boolean reportDriverConfig) { + public void populateControlConnectionOptions( + Map startupOptions, boolean scyllaDb) { // Configuration reporting is a best-effort diagnostic aid: it runs on the connection // initialization path, so any failure here (a bad config read, a misbehaving policy while // introspecting, a serialization error) must be swallowed rather than allowed to break the - // connection — which would prevent the session from establishing or reconnecting. + // connection — which would prevent the session from establishing or reconnecting. Also catches + // InternalError specifically: customPolicy() calls getClass().getSimpleName() on arbitrary + // user-supplied policy objects, which has a documented JDK edge case throwing InternalError for + // certain synthetic classes. Deliberately not a bare `Error` — that would also swallow + // OutOfMemoryError/StackOverflowError, masking a real JVM-level failure instead of this one + // narrow, documented case. try { if (!isEnabled()) { return; } - // SESSION_ID on every connection so the server can group a session's connections. - startupOptions.put(SESSION_ID_KEY, sessionId.toString()); - // DRIVER_CONFIG blob only on the control connection. - if (reportDriverConfig) { - String json = buildJson(); - if (json != null) { - startupOptions.put(DRIVER_CONFIG_KEY, json); - } + String json = buildJson(scyllaDb); + if (json != null) { + startupOptions.put(DRIVER_CONFIG_KEY, json); } - } catch (RuntimeException e) { - LOG.warn( - "Error while building the driver configuration report; skipping driver config reporting", - e); + } catch (InternalError | RuntimeException e) { + LOG.warn("Error while building the driver configuration report; skipping DRIVER_CONFIG", e); } } + // Read on every control-connection initialization rather than cached, so that a configuration + // reload takes effect on the next (re)connect. The fallback mirrors the reference.conf default, + // so that a configuration omitting the option behaves like the shipped one. private boolean isEnabled() { return context .getConfig() .getDefaultProfile() - .getBoolean(DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, false); + .getBoolean(DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, true); } /** * Builds the compact, single-line JSON configuration report. * - *

Stage 1 emits only the schema {@code version}; the individual configuration groups are - * populated in {@link #populateConfig(ObjectNode, DriverExecutionProfile)} in a later stage. + *

Relies on the policy/generator {@code LazyReference}s (reconnection, retry, speculative + * execution, load balancing, timestamp generator, SSL engine factory) already being resolved by + * the time this runs, which holds today because session bootstrap eagerly forces them before + * {@code ProtocolInitHandler} sends the first {@code STARTUP}. That ordering isn't enforced by + * this class; a future change to session bootstrap that defers one of those references could make + * this the first caller to resolve it, from a Netty event-loop thread mid-{@code STARTUP} build. + * + * @param scyllaDb whether the control connection is talking to ScyllaDB; see {@link + * DriverConfigReporter#populateControlConnectionOptions}. */ - protected String buildJson() { + String buildJson(boolean scyllaDb) { ObjectNode root = OBJECT_MAPPER.createObjectNode(); root.put("version", SCHEMA_VERSION); - populateConfig(root, context.getConfig().getDefaultProfile()); + populateConfig(root, context.getConfig().getDefaultProfile(), scyllaDb); try { return OBJECT_MAPPER.writeValueAsString(root); } catch (JsonProcessingException e) { @@ -121,11 +161,317 @@ protected String buildJson() { } /** - * Populates the configuration groups onto the report root. Placeholder in Stage 1; Stage 2 fills - * in {@code connection}, {@code socket}, the policy groups, {@code query_defaults}, {@code tls}, - * etc. + * Populates the configuration groups onto the report root, from the default execution profile + * plus the context's policies. Each group follows the cross-driver schema; a key the Java driver + * has no equivalent for is omitted rather than reported as {@code null}. + */ + private void populateConfig(ObjectNode root, DriverExecutionProfile config, boolean scyllaDb) { + root.set("connection", connection(config)); + root.set("socket", socket(config)); + root.set("control-plane", controlPlane(config, scyllaDb)); + root.set("reconnection-policy", reconnectionPolicy(config)); + root.set("retry-policy", retryPolicy()); + // No null variant in the schema for this group: omitted entirely when there is none. + ObjectNode specExec = speculativeExecutionPolicy(config); + if (specExec != null) { + root.set("speculative-execution-policy", specExec); + } + root.set("load-balancing-policy", loadBalancingPolicy(config)); + root.set("node-location-preference", nodeLocationPreference(config)); + root.set("connection-pool", connectionPool(config)); + root.set("query-defaults", queryDefaults(config)); + root.set("tls", tls()); + } + + private ObjectNode connection(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + ObjectNode connect = OBJECT_MAPPER.createObjectNode(); + connect.put( + "timeout-ms", + config.getDuration(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT).toMillis()); + n.set("connect", connect); + // The Java driver has no socket-level read/write timeouts, and connection.heartbeat is a + // reserved empty placeholder in this schema version (no slot for HEARTBEAT_INTERVAL/TIMEOUT + // yet) — all three are omitted entirely rather than reported as empty/null. + return n; + } + + private ObjectNode socket(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("tcp-no-delay", config.getBoolean(DefaultDriverOption.SOCKET_TCP_NODELAY, true)); + // keep-alive and reuse-address are unset by default; the driver leaves the socket option + // untouched, so the effective value is the JDK/OS default — false for both SO_KEEPALIVE and + // (client-socket) SO_REUSEADDR. The schema requires both keys, so they are always emitted. + n.put("keep-alive", config.getBoolean(DefaultDriverOption.SOCKET_KEEP_ALIVE, false)); + n.put("reuse-address", config.getBoolean(DefaultDriverOption.SOCKET_REUSE_ADDRESS, false)); + if (config.isDefined(DefaultDriverOption.SOCKET_LINGER_INTERVAL)) { + ObjectNode linger = OBJECT_MAPPER.createObjectNode(); + linger.put("interval-s", config.getInt(DefaultDriverOption.SOCKET_LINGER_INTERVAL)); + n.set("linger", linger); + } + if (config.isDefined(DefaultDriverOption.SOCKET_RECEIVE_BUFFER_SIZE)) { + ObjectNode receiveBuffer = OBJECT_MAPPER.createObjectNode(); + receiveBuffer.put( + "size-bytes", config.getInt(DefaultDriverOption.SOCKET_RECEIVE_BUFFER_SIZE)); + n.set("receive-buffer", receiveBuffer); + } + if (config.isDefined(DefaultDriverOption.SOCKET_SEND_BUFFER_SIZE)) { + ObjectNode sendBuffer = OBJECT_MAPPER.createObjectNode(); + sendBuffer.put("size-bytes", config.getInt(DefaultDriverOption.SOCKET_SEND_BUFFER_SIZE)); + n.set("send-buffer", sendBuffer); + } + return n; + } + + private ObjectNode controlPlane(DriverExecutionProfile config, boolean scyllaDb) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + + ObjectNode timeout = OBJECT_MAPPER.createObjectNode(); + timeout.put( + "client-side-ms", + config.getDuration(DefaultDriverOption.CONTROL_CONNECTION_TIMEOUT).toMillis()); + if (scyllaDb) { + // ScyllaDB only: CassandraSchemaQueries adds a "USING TIMEOUT ms" clause (this same + // value) to every schema query it runs, making METADATA_SCHEMA_REQUEST_TIMEOUT a genuine + // server-side timeout on this backend. Genuine Cassandra never gets that clause, so the + // field is omitted there (not applicable). + timeout.put( + "server-side-ms", + config.getDuration(DefaultDriverOption.METADATA_SCHEMA_REQUEST_TIMEOUT).toMillis()); + } + ObjectNode systemQueries = OBJECT_MAPPER.createObjectNode(); + systemQueries.set("timeout", timeout); + n.set("system-queries", systemQueries); + + ObjectNode schemaAgreement = OBJECT_MAPPER.createObjectNode(); + schemaAgreement.put( + "timeout-ms", + config.getDuration(DefaultDriverOption.CONTROL_CONNECTION_AGREEMENT_TIMEOUT).toMillis()); + n.set("schema-agreement", schemaAgreement); + + return n; + } + + private ObjectNode reconnectionPolicy(DriverExecutionProfile config) { + ReconnectionPolicy policy = context.getReconnectionPolicy(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // Exact-class checks, not instanceof: none of these built-ins are final, so a user subclass + // (e.g. to tweak one method) must fall through to the "custom" branch below rather than be + // misreported as the unmodified built-in. + if (policy.getClass() == ExponentialReconnectionPolicy.class) { + n.put("type", "exponential"); + n.put("base-ms", config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); + n.put("max-ms", config.getDuration(DefaultDriverOption.RECONNECTION_MAX_DELAY).toMillis()); + // Java's built-in reconnection policies are unbounded: max-attempts is omitted. + } else if (policy.getClass() == ConstantReconnectionPolicy.class) { + n.put("type", "constant"); + n.put("delay-ms", config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); + } else { + customPolicy(n, policy); + } + return n; + } + + private ObjectNode retryPolicy() { + RetryPolicy policy = context.getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // Exact-class check, not instanceof: DefaultRetryPolicy/ConsistencyDowngradingRetryPolicy are + // not final, so a user subclass must fall through to "custom" rather than be misreported. + if (policy.getClass() == DefaultRetryPolicy.class) { + n.put("type", "standard-error-aware"); + // No configurable backoff: omitted. + } else if (policy.getClass() == ConsistencyDowngradingRetryPolicy.class) { + n.put("type", "downgrading-consistency"); + // No configurable backoff: omitted. + } else { + customPolicy(n, policy); + } + return n; + } + + /** + * Returns {@code null} when there is no speculative execution policy to report, in which case the + * whole group is omitted from the report (the schema has no null variant for it). + */ + private ObjectNode speculativeExecutionPolicy(DriverExecutionProfile config) { + SpeculativeExecutionPolicy policy = + context.getSpeculativeExecutionPolicy(DriverExecutionProfile.DEFAULT_NAME); + // Exact-class checks, not instanceof: neither built-in is final. + if (policy.getClass() == NoSpeculativeExecutionPolicy.class) { + return null; + } + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy.getClass() == ConstantSpeculativeExecutionPolicy.class) { + n.put("type", "constant"); + n.put("max-executions", config.getInt(DefaultDriverOption.SPECULATIVE_EXECUTION_MAX)); + n.put( + "delay-ms", + config.getDuration(DefaultDriverOption.SPECULATIVE_EXECUTION_DELAY).toMillis()); + } else { + customPolicy(n, policy); + } + return n; + } + + private ObjectNode loadBalancingPolicy(DriverExecutionProfile config) { + LoadBalancingPolicy policy = + context.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME); + Class policyClass = policy.getClass(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // DcInferringLoadBalancingPolicy extends DefaultLoadBalancingPolicy, overriding only how the + // local DC is discovered; DseLoadBalancingPolicy/DseDcInferringLoadBalancingPolicy are + // deprecated, behavior-identical aliases of the two ("equivalent to DefaultLoadBalancingPolicy, + // which should now be used instead" per their own javadoc). All three built-in policies are + // always token-aware and unconditionally shuffle replicas whenever more than one is available + // (neither has a config option to disable shuffling) and honor the same DC-failover option; + // only "type" and "latency-awareness" differ, so those two are resolved per class below and the + // shared fields are written once. Exact-class checks (not instanceof) so an actual user + // subclass of any of these still falls through to "custom" below. + boolean isDcInferring = + policyClass == DcInferringLoadBalancingPolicy.class + || policyClass == DseDcInferringLoadBalancingPolicy.class; + String type; + boolean latencyAwareness; + if (isDcInferring + || policyClass == DefaultLoadBalancingPolicy.class + || policyClass == DseLoadBalancingPolicy.class) { + type = isDcInferring ? "dc-inferring" : "default"; + // No classic latency-percentile host ordering; the closest available signal is slow-replica + // avoidance (a busy/health-based reorder of already-selected replicas), on by default. + latencyAwareness = + config.getBoolean(DefaultDriverOption.LOAD_BALANCING_POLICY_SLOW_AVOIDANCE, true); + } else if (policyClass == BasicLoadBalancingPolicy.class) { + type = "basic"; + // Unlike DefaultLoadBalancingPolicy, BasicLoadBalancingPolicy has no slow-replica-avoidance + // mechanism at all. + latencyAwareness = false; + } else { + customPolicy(n, policy); + return n; + } + n.put("type", type); + n.put("token-aware", true); + n.put("shuffle", true); + n.put( + "dc-failover", + config.getInt(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_MAX_NODES_PER_REMOTE_DC, 0) + > 0); + n.put("latency-awareness", latencyAwareness); + return n; + } + + /** + * Session-level datacenter/rack preference. Java has no session-level locality API separate from + * the load balancing policy, so this is sourced from the same places the (default) load balancing + * policy itself reads locality from: the local DC can be set either programmatically via {@link + * com.datastax.oss.driver.api.core.session.SessionBuilder#withLocalDatacenter} (which takes + * precedence, mirroring {@code OptionalLocalDcHelper}) or via config; the local rack has no + * programmatic override and is config-only. + * + *

Known limitation: when neither is set, this reports {@code dc-auto} for the entire + * lifetime of the session, not just "not yet known for this particular report" — including on + * later control-connection reconnects, long after {@code DcInferringLoadBalancingPolicy} has + * resolved a real DC. That resolved value ({@code BasicLoadBalancingPolicy#getLocalDatacenter()}) + * is {@code protected}, on an internal class in a different package, and not exposed anywhere on + * the public {@link LoadBalancingPolicy} interface or {@link InternalDriverContext}; reporting it + * would need new public API surface, out of scope for this reporter. */ - protected void populateConfig(ObjectNode root, DriverExecutionProfile config) { - // Stage 2: populate configuration groups from `config` and the context's policies. + private ObjectNode nodeLocationPreference(DriverExecutionProfile config) { + String localDc = context.getLocalDatacenter(DriverExecutionProfile.DEFAULT_NAME); + if (localDc == null && config.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) { + localDc = config.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER); + } + String localRack = + config.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK) + ? config.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK) + : null; + + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (localDc != null && localRack != null) { + n.put("type", "rack"); + n.put("local-dc", localDc); + n.put("local-rack", localRack); + } else if (localDc != null) { + n.put("type", "dc"); + n.put("local-dc", localDc); + } else if (localRack != null) { + // DC isn't explicit (DefaultLoadBalancingPolicy will infer one from the first contacted + // node), but rack was configured explicitly on its own: report rack-auto so the known rack + // isn't silently dropped. local-dc is omitted since it isn't known yet at report time. + n.put("type", "rack-auto"); + n.put("local-rack", localRack); + } else { + // DC is inferred from the first contacted node; not known yet at control-connection-init + // report time, so local-dc is omitted. + n.put("type", "dc-auto"); + } + return n; + } + + private ObjectNode connectionPool(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("type", "host"); + n.put( + "desired-connections-count", config.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE)); + if (config.isDefined(DefaultDriverOption.CONNECTION_MAX_REQUESTS)) { + ObjectNode connection = OBJECT_MAPPER.createObjectNode(); + connection.put("max-requests", config.getInt(DefaultDriverOption.CONNECTION_MAX_REQUESTS)); + n.set("connection", connection); + } + ObjectNode shardAware = OBJECT_MAPPER.createObjectNode(); + shardAware.put( + "enabled", + config.getBoolean(DefaultDriverOption.CONNECTION_ADVANCED_SHARD_AWARENESS_ENABLED, true)); + n.set("shard-aware", shardAware); + return n; + } + + private ObjectNode queryDefaults(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + int pageSize = config.getInt(DefaultDriverOption.REQUEST_PAGE_SIZE); + if (pageSize > 0) { + ObjectNode page = OBJECT_MAPPER.createObjectNode(); + page.put("size", pageSize); + n.set("page", page); + } + // pageSize <= 0 means paging is unbounded: the "page" group is omitted entirely (the schema + // has no "unbounded" sentinel). + n.put("consistency", config.getString(DefaultDriverOption.REQUEST_CONSISTENCY)); + if (config.isDefined(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY)) { + n.put("serial-consistency", config.getString(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY)); + } + n.put("idempotence", config.getBoolean(DefaultDriverOption.REQUEST_DEFAULT_IDEMPOTENCE)); + // Client-side timestamps are assigned unless the server-side generator is configured. + n.put( + "client-timestamps", + !(context.getTimestampGenerator() instanceof ServerSideTimestampGenerator)); + ObjectNode request = OBJECT_MAPPER.createObjectNode(); + request.put("timeout-ms", config.getDuration(DefaultDriverOption.REQUEST_TIMEOUT).toMillis()); + n.set("request", request); + return n; + } + + private ObjectNode tls() { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // Report the factory's real hostname-validation state, not the SSL_HOSTNAME_VALIDATION config + // option: that option only governs the built-in DefaultSslEngineFactory. A factory supplied via + // SessionBuilder.withSslContext(...) (ProgrammaticSslEngineFactory) validates only if + // explicitly + // asked to (default off) regardless of that option, so reading the option here would falsely + // report validation as on when it isn't. + Optional factory = context.getSslEngineFactory(); + n.put("enabled", factory.isPresent()); + n.put( + "hostname-verification", + factory.map(SslEngineFactory::requireHostnameValidation).orElse(false)); + return n; + } + + private void customPolicy(ObjectNode node, Object policy) { + node.put("type", "custom"); + // getSimpleName() is empty for an anonymous class (a common way to supply a one-off policy); + // fall back to the full (binary) name so the policy is still identifiable. + String name = policy.getClass().getSimpleName(); + node.put("name", name.isEmpty() ? policy.getClass().getName() : name); } } 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 3d1d5b82b87..2983ef0787f 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 @@ -361,6 +361,10 @@ public DefaultDriverContext( /** * Returns the options to send in a Startup message. * + *

Called once per session (the result is held by a {@code LazyReference} and copied into every + * connection's {@code STARTUP}), which is what makes the {@link + * StartupOptionsBuilder#SESSION_ID_KEY SESSION_ID} it contains stable for the whole session. + * * @see #getStartupOptions() */ protected Map buildStartupOptions() { diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java index d614793e9d0..236b2e5da99 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java @@ -20,34 +20,32 @@ import java.util.Map; /** - * Adds the client-configuration-reporting entries to a connection's CQL {@code STARTUP} options, so - * ScyllaDB can store them in {@code system.clients.client_options} and operators can inspect a - * client's effective driver settings while investigating incidents. + * Adds the {@code DRIVER_CONFIG} entry to the control connection's CQL {@code STARTUP} options, so + * ScyllaDB can store it in {@code system.clients.client_options} and operators can inspect the + * driver's effective settings while investigating incidents. * - *

Two entries are produced, both governed by {@code advanced.driver-config-reporting.enabled}: + *

The blob describes the whole session, so only the control connection carries it — pooled + * connections are correlated back to it through the {@link StartupOptionsBuilder#SESSION_ID_KEY + * SESSION_ID} startup option, which the driver sends on every connection unconditionally and + * independently of this reporter. * - *

+ *

Governed by {@code advanced.driver-config-reporting.enabled} (enabled by default). */ public interface DriverConfigReporter { /** - * Adds the reporting entries to the given startup options: {@code SESSION_ID} on every - * connection, plus {@code DRIVER_CONFIG} when {@code reportDriverConfig} is true (the control - * connection). Does nothing when configuration reporting is disabled. + * Adds the {@code DRIVER_CONFIG} blob to the given startup options, unless configuration + * reporting is disabled. * - *

Called from the protocol-initialization handler for every connection. + *

Called from the protocol-initialization handler for the control connection only. * *

Implementations must not throw: this runs on the connection initialization path, so a * failure to build the report must be swallowed (and logged) rather than propagated, otherwise it * would prevent the session from establishing or reconnecting. * - * @param reportDriverConfig whether this connection should also carry the full {@code - * DRIVER_CONFIG} blob; true only for the control connection. + * @param scyllaDb whether this connection is talking to ScyllaDB (as opposed to a generic + * Cassandra server), so the report can reflect server-side behavior that only applies on that + * backend (e.g. the {@code USING TIMEOUT} clause added to schema queries). */ - void populateStartupOptions(Map startupOptions, boolean reportDriverConfig); + void populateControlConnectionOptions(Map startupOptions, boolean scyllaDb); } 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..dd3a0307a79 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 @@ -37,8 +37,20 @@ public class StartupOptionsBuilder { public static final String APPLICATION_VERSION_KEY = "APPLICATION_VERSION"; public static final String CLIENT_ID_KEY = "CLIENT_ID"; + /** + * STARTUP option key under which the session's identifier is sent, so that the server can group + * all of a session's connections (and correlate them with the configuration that the control + * connection reports under {@code DRIVER_CONFIG}). + * + *

This is an innate driver behavior: the option is sent on every connection, unconditionally. + * In particular it is not governed by {@code advanced.driver-config-reporting.enabled}, + * which only decides whether the control connection also reports the configuration itself. + */ + public static final String SESSION_ID_KEY = "SESSION_ID"; + protected final InternalDriverContext context; private UUID clientId; + private UUID sessionId; private String applicationName; private String applicationVersion; @@ -84,9 +96,9 @@ public StartupOptionsBuilder withApplicationVersion(@Nullable String application * *

The default set of options are built here and include {@link * com.datastax.oss.protocol.internal.request.Startup#COMPRESSION_KEY} (if the context passed in - * has a compressor/algorithm set), and the driver's {@link #DRIVER_NAME_KEY} and {@link - * #DRIVER_VERSION_KEY}. The {@link com.datastax.oss.protocol.internal.request.Startup} - * constructor will add {@link + * has a compressor/algorithm set), the driver's {@link #DRIVER_NAME_KEY} and {@link + * #DRIVER_VERSION_KEY}, and the {@link #SESSION_ID_KEY}. The {@link + * com.datastax.oss.protocol.internal.request.Startup} constructor will add {@link * com.datastax.oss.protocol.internal.request.Startup#CQL_VERSION_KEY}. * * @return Map of Startup Options. @@ -94,7 +106,7 @@ public StartupOptionsBuilder withApplicationVersion(@Nullable String application public Map build() { DriverExecutionProfile config = context.getConfig().getDefaultProfile(); - NullAllowingImmutableMap.Builder builder = NullAllowingImmutableMap.builder(3); + NullAllowingImmutableMap.Builder builder = NullAllowingImmutableMap.builder(4); // add compression (if configured) and driver name and version String compressionAlgorithm = context.getCompressor().algorithm(); if (compressionAlgorithm != null && !compressionAlgorithm.trim().isEmpty()) { @@ -102,6 +114,17 @@ public Map build() { } builder.put(DRIVER_NAME_KEY, getDriverName()).put(DRIVER_VERSION_KEY, getDriverVersion()); + // Identifier of this session, sent on every connection so the server can group them. Not + // derived from the (user-settable, Insights-oriented) CLIENT_ID below, so that it is guaranteed + // unique per session as the grouping key requires. Generated lazily here rather than eagerly in + // a field initializer, mirroring clientId; DefaultDriverContext builds the startup options + // exactly once per session (LazyReference), which is what makes the value stable across all of + // the session's connections, including reconnects. + if (sessionId == null) { + sessionId = Uuids.random(); + } + builder.put(SESSION_ID_KEY, sessionId.toString()); + // Add Insights entries, falling back to generation / config if no programmatic values provided: if (clientId == null) { clientId = Uuids.random(); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/CassandraSchemaQueries.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/CassandraSchemaQueries.java index f909a0cb387..59a5a50c9b6 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/CassandraSchemaQueries.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/CassandraSchemaQueries.java @@ -171,7 +171,10 @@ private void executeOnAdminExecutor() { } protected boolean shouldApplyUsingTimeout() { - // We use non-null sharding info as a proxy check for cluster being a ScyllaDB cluster + // We use non-null sharding info as a proxy check for cluster being a ScyllaDB cluster. + // The same check (independently, on the control channel) backs the "scyllaDb" signal that + // DefaultDriverConfigReporter uses to decide whether to report a server-side USING TIMEOUT + // value; see ProtocolInitHandler's STARTUP case. return (channel.getShardingInfo() != null); } 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 343d3f9e4e7..6d3fb0bdc35 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 @@ -137,6 +137,11 @@ public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) { return engine; } + @Override + public boolean requireHostnameValidation() { + return requireHostnameValidation; + } + protected SSLContext buildContext(DriverExecutionProfile config) throws Exception { if (config.isDefined(DefaultDriverOption.SSL_KEYSTORE_PATH) || config.isDefined(DefaultDriverOption.SSL_TRUSTSTORE_PATH)) { 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 4d2cb69fbfc..c58024a98ff 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 @@ -92,6 +92,12 @@ private int getFakePort(String sniServerName) { return FAKE_PORT_OFFSET + fakePorts.indexOf(sniServerName); } + @Override + public boolean requireHostnameValidation() { + // SNI connections always set the "HTTPS" endpoint identification algorithm above. + return true; + } + @Override public void close() { // nothing to do diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf index 8a3a444319e..5c7c4972c58 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1206,21 +1206,26 @@ datastax-java-driver { advanced.driver-config-reporting { - # Whether the driver reports its effective configuration to ScyllaDB at connection time. + # Whether the driver reports its effective configuration to the cluster at connection time. # - # When true, the driver adds two entries to the CQL STARTUP options, which ScyllaDB stores in - # system.clients.client_options so operators can inspect driver settings while investigating - # incidents: a SESSION_ID on every connection (so the server can group a session's connections) - # and a compact JSON payload under the DRIVER_CONFIG key on the control connection only. At this - # stage the DRIVER_CONFIG payload carries only schema-version metadata ({"version":1}); reporting - # of the effective configuration fields is planned for a later stage. When false, neither entry - # is sent and there is no change on the wire. + # When true, the control connection adds a compact JSON payload under the DRIVER_CONFIG key to + # its CQL STARTUP options, which the server stores in its client-connection system table + # (system.clients on ScyllaDB, system_views.clients on Cassandra 4.1+) so operators can inspect + # driver settings while investigating incidents. It describes the effective configuration of the + # driver's default execution profile (connection/socket settings, timeouts, + # retry/reconnection/speculative-execution/load-balancing policies, connection pooling, query + # defaults, and TLS). Only the control connection sends it, since it describes the whole + # session. When false, DRIVER_CONFIG is not sent. + # + # Note that this option does not govern the SESSION_ID startup option, which the driver always + # sends on every connection so that the server can group a session's connections (and correlate + # them with the configuration reported here). # # Required: no # Modifiable at runtime: yes, the new value will be used for connections initialized after the change. # Overridable in a profile: no - # Default: false - enabled = false + # Default: true + enabled = true } diff --git a/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java b/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java index 9e4556e528d..81df53af9a3 100644 --- a/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java +++ b/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java @@ -80,6 +80,8 @@ private void assertDefaultStartupOptions(Startup startup) { Version version = Version.parse(startup.options.get(StartupOptionsBuilder.DRIVER_VERSION_KEY)); assertThat(version).isEqualTo(Session.OSS_DRIVER_COORDINATES.getVersion()); assertThat(startup.options).containsKey(StartupOptionsBuilder.CLIENT_ID_KEY); + // SESSION_ID is innate and must survive on the DSE path too. + assertThat(startup.options).containsKey(StartupOptionsBuilder.SESSION_ID_KEY); } @Test diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java index 2780d5bdec9..15ed28d7a03 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java @@ -141,8 +141,8 @@ public void setup() throws InterruptedException { when(context.getEventBus()).thenReturn(eventBus); when(context.getWriteCoalescer()).thenReturn(new PassThroughWriteCoalescer(null)); when(context.getCompressor()).thenReturn(compressor); - // The init handler consults the config reporter for every connection; default to a no-op. - when(context.getDriverConfigReporter()).thenReturn((startupOptions, reportDriverConfig) -> {}); + // The init handler consults the config reporter for the control connection; default to a no-op. + when(context.getDriverConfigReporter()).thenReturn((startupOptions, scyllaDb) -> {}); // Start local server ServerBootstrap serverBootstrap = diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java index a7051ac466d..1a9633b5049 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java @@ -19,6 +19,7 @@ import static com.datastax.oss.driver.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -42,9 +43,12 @@ import com.datastax.oss.driver.internal.core.ProtocolVersionRegistry; import com.datastax.oss.driver.internal.core.TestResponses; import com.datastax.oss.driver.internal.core.context.DefaultDriverConfigReporter; +import com.datastax.oss.driver.internal.core.context.DriverConfigReporter; import com.datastax.oss.driver.internal.core.context.InternalDriverContext; +import com.datastax.oss.driver.internal.core.context.StartupOptionsBuilder; import com.datastax.oss.driver.internal.core.metadata.TestNodeFactory; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList; +import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap; import com.datastax.oss.protocol.internal.Frame; import com.datastax.oss.protocol.internal.ProtocolConstants; import com.datastax.oss.protocol.internal.ProtocolConstants.ErrorCode; @@ -58,6 +62,7 @@ import com.datastax.oss.protocol.internal.response.Authenticate; import com.datastax.oss.protocol.internal.response.Error; import com.datastax.oss.protocol.internal.response.Ready; +import com.datastax.oss.protocol.internal.response.Supported; import com.datastax.oss.protocol.internal.response.result.SetKeyspace; import com.datastax.oss.protocol.internal.util.Bytes; import io.netty.channel.ChannelFuture; @@ -69,6 +74,7 @@ import java.util.Map; import java.util.Optional; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Before; import org.junit.Test; import org.mockito.Mock; @@ -103,9 +109,9 @@ public void setup() { when(defaultProfile.getDuration(DefaultDriverOption.HEARTBEAT_INTERVAL)) .thenReturn(Duration.ofSeconds(30)); when(internalDriverContext.getProtocolVersionRegistry()).thenReturn(protocolVersionRegistry); - // The init handler consults the config reporter for every connection; default to a no-op. + // The init handler consults the config reporter for the control connection; default to a no-op. when(internalDriverContext.getDriverConfigReporter()) - .thenReturn((startupOptions, reportDriverConfig) -> {}); + .thenReturn((startupOptions, scyllaDb) -> {}); channel .pipeline() @@ -157,21 +163,17 @@ public void should_initialize() { assertThat(connectFuture).isSuccess(); } - // Mirrors the real reporter: SESSION_ID on every connection, DRIVER_CONFIG only when asked. + // Mirrors the real reporter, which only ever sees the control connection. private void stubConfigReporter() { when(internalDriverContext.getDriverConfigReporter()) .thenReturn( - (startupOptions, reportDriverConfig) -> { - startupOptions.put(DefaultDriverConfigReporter.SESSION_ID_KEY, "test-session-id"); - if (reportDriverConfig) { + (startupOptions, scyllaDb) -> startupOptions.put( - DefaultDriverConfigReporter.DRIVER_CONFIG_KEY, "{\"version\":1}"); - } - }); + DefaultDriverConfigReporter.DRIVER_CONFIG_KEY, "{\"version\":1}")); } @Test - public void should_report_session_id_and_driver_config_on_control_connection() { + public void should_report_driver_config_on_control_connection() { stubConfigReporter(); channel .pipeline() @@ -191,13 +193,13 @@ public void should_report_session_id_and_driver_config_on_control_connection() { Frame requestFrame = readOutboundFrame(); assertThat(requestFrame.message).isInstanceOf(Startup.class); Startup startup = (Startup) requestFrame.message; - assertThat(startup.options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); assertThat(startup.options).containsKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); } @Test - public void should_report_session_id_but_not_driver_config_on_pool_connection() { - stubConfigReporter(); + public void should_not_consult_the_config_reporter_on_pool_connection() { + DriverConfigReporter reporter = mock(DriverConfigReporter.class); + when(internalDriverContext.getDriverConfigReporter()).thenReturn(reporter); channel .pipeline() .addLast( @@ -217,8 +219,36 @@ public void should_report_session_id_but_not_driver_config_on_pool_connection() Frame requestFrame = readOutboundFrame(); assertThat(requestFrame.message).isInstanceOf(Startup.class); Startup startup = (Startup) requestFrame.message; - assertThat(startup.options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); assertThat(startup.options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + verify(reporter, never()).populateControlConnectionOptions(any(), anyBoolean()); + } + + @Test + public void should_pass_session_id_from_the_session_startup_options_to_every_connection() { + // SESSION_ID is not the reporter's business: it comes from the session-wide startup options, so + // it reaches pool connections (reportConfig = false) as well. + when(internalDriverContext.getStartupOptions()) + .thenReturn(ImmutableMap.of(StartupOptionsBuilder.SESSION_ID_KEY, "test-session-id")); + channel + .pipeline() + .addLast( + ChannelFactory.INIT_HANDLER_NAME, + new ProtocolInitHandler( + internalDriverContext, + DefaultProtocolVersion.V4, + null, + END_POINT, + DriverChannelOptions.DEFAULT, + heartbeatHandler, + false)); + + channel.connect(new InetSocketAddress("localhost", 9042)); + + Frame requestFrame = readOutboundFrame(); + assertThat(requestFrame.message).isInstanceOf(Startup.class); + Startup startup = (Startup) requestFrame.message; + assertThat(startup.options) + .containsEntry(StartupOptionsBuilder.SESSION_ID_KEY, "test-session-id"); } @Test @@ -267,6 +297,81 @@ public void should_query_supported_options() { assertThat(connectFuture).isSuccess(); } + @Test + public void should_report_scylladb_true_when_sharding_info_present() { + AtomicBoolean capturedScyllaDb = new AtomicBoolean(); + when(internalDriverContext.getDriverConfigReporter()) + .thenReturn((startupOptions, scyllaDb) -> capturedScyllaDb.set(scyllaDb)); + channel + .pipeline() + .addLast( + ChannelFactory.INIT_HANDLER_NAME, + new ProtocolInitHandler( + internalDriverContext, + DefaultProtocolVersion.V4, + null, + END_POINT, + // Only the control connection reports the config, so only it computes scyllaDb. + DriverChannelOptions.builder().reportConfig(true).build(), + heartbeatHandler, + true)); + + channel.connect(new InetSocketAddress("localhost", 9042)); + + Frame requestFrame = readOutboundFrame(); + assertThat(requestFrame.message).isInstanceOf(Options.class); + + // Simulate a SUPPORTED response carrying the five ScyllaDB sharding-info keys + // ShardingInfo.parseShardingInfo() requires (see ShardingInfo.java:112-128). + Map> shardingOptions = + ImmutableMap.>builder() + .put("SCYLLA_SHARD", ImmutableList.of("0")) + .put("SCYLLA_NR_SHARDS", ImmutableList.of("4")) + .put( + "SCYLLA_PARTITIONER", + ImmutableList.of("org.apache.cassandra.dht.Murmur3Partitioner")) + .put("SCYLLA_SHARDING_ALGORITHM", ImmutableList.of("biased-token-round-robin")) + .put("SCYLLA_SHARDING_IGNORE_MSB", ImmutableList.of("12")) + .build(); + writeInboundFrame(requestFrame, new Supported(shardingOptions)); + + requestFrame = readOutboundFrame(); + assertThat(requestFrame.message).isInstanceOf(Startup.class); + assertThat(capturedScyllaDb.get()).isTrue(); + } + + @Test + public void should_report_scylladb_false_when_sharding_info_absent() { + AtomicBoolean capturedScyllaDb = new AtomicBoolean(true); + when(internalDriverContext.getDriverConfigReporter()) + .thenReturn((startupOptions, scyllaDb) -> capturedScyllaDb.set(scyllaDb)); + channel + .pipeline() + .addLast( + ChannelFactory.INIT_HANDLER_NAME, + new ProtocolInitHandler( + internalDriverContext, + DefaultProtocolVersion.V4, + null, + END_POINT, + // Only the control connection reports the config, so only it computes scyllaDb. + DriverChannelOptions.builder().reportConfig(true).build(), + heartbeatHandler, + true)); + + channel.connect(new InetSocketAddress("localhost", 9042)); + + Frame requestFrame = readOutboundFrame(); + assertThat(requestFrame.message).isInstanceOf(Options.class); + + // No sharding-info keys: ShardingInfo.parseShardingInfo(...) returns null. + writeInboundFrame(requestFrame, TestResponses.supportedResponse("mock_key", "mock_value")); + + requestFrame = readOutboundFrame(); + assertThat(requestFrame.message).isInstanceOf(Startup.class); + assertThat(capturedScyllaDb.get()).isFalse(); + } + @Test public void should_add_heartbeat_handler_to_pipeline_on_success() { ProtocolInitHandler protocolInitHandler = diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java index b303e6db5a5..f7948a34cb0 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java @@ -21,121 +21,959 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import com.datastax.dse.driver.internal.core.loadbalancing.DseDcInferringLoadBalancingPolicy; +import com.datastax.dse.driver.internal.core.loadbalancing.DseLoadBalancingPolicy; import com.datastax.oss.driver.api.core.config.DefaultDriverOption; 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.config.OptionsMap; +import com.datastax.oss.driver.api.core.config.TypedDriverOption; +import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy; +import com.datastax.oss.driver.api.core.context.DriverContext; +import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy; +import com.datastax.oss.driver.api.core.retry.RetryPolicy; +import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy; +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.time.TimestampGenerator; +import com.datastax.oss.driver.internal.core.connection.ConstantReconnectionPolicy; +import com.datastax.oss.driver.internal.core.connection.ExponentialReconnectionPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.BasicLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DcInferringLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.retry.ConsistencyDowngradingRetryPolicy; +import com.datastax.oss.driver.internal.core.retry.DefaultRetryPolicy; +import com.datastax.oss.driver.internal.core.specex.ConstantSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.specex.NoSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.time.ServerSideTimestampGenerator; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; +import com.networknt.schema.JsonSchema; +import com.networknt.schema.JsonSchemaFactory; +import com.networknt.schema.SpecVersion; +import com.networknt.schema.ValidationMessage; +import java.io.InputStream; +import java.time.Duration; import java.util.HashMap; import java.util.Map; -import java.util.UUID; +import java.util.Optional; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Supplier; +import javax.net.ssl.SSLContext; import org.junit.Before; import org.junit.Test; +// Many tests below use mock(SomeBuiltinPolicy.class) and assert the reporter recognizes it as that +// exact built-in (not "custom"). This relies on Mockito 5's default inline mock maker returning an +// object whose getClass() is the literal mocked class rather than a generated subclass (verified +// empirically for this project's Mockito version); a return to subclass-based mocking would make +// every exact-class branch under test here fall through to "custom" instead. public class DefaultDriverConfigReporterTest { - private InternalDriverContext context; - private DriverExecutionProfile profile; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + // The normative v1 JSON Schema from the design doc, shipped verbatim as a test resource. Loaded + // once and pinned to draft 2020-12 (its declared $schema); its internal "#/$defs/..." refs + // resolve locally, so validation needs no network access. + private static final JsonSchema SCHEMA = loadSchema(); + + private static JsonSchema loadSchema() { + try (InputStream in = + DefaultDriverConfigReporterTest.class.getResourceAsStream( + "/config/driver-config-report-v1.schema.json")) { + return JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V202012) + .getSchema(MAPPER.readTree(in)); + } catch (Exception e) { + throw new AssertionError("Cannot load the DRIVER_CONFIG v1 JSON Schema resource", e); + } + } + + // ---- Fixtures for the gating / fail-safe tests (bare mock profile) ---- + private InternalDriverContext mockContext; + private DriverExecutionProfile mockProfile; private DefaultDriverConfigReporter reporter; @Before public void setup() { - context = mock(InternalDriverContext.class); + mockContext = mock(InternalDriverContext.class); DriverConfig config = mock(DriverConfig.class); - profile = mock(DriverExecutionProfile.class); - when(context.getConfig()).thenReturn(config); - when(config.getDefaultProfile()).thenReturn(profile); - reporter = new DefaultDriverConfigReporter(context); + mockProfile = mock(DriverExecutionProfile.class); + when(mockContext.getConfig()).thenReturn(config); + when(config.getDefaultProfile()).thenReturn(mockProfile); + reporter = new DefaultDriverConfigReporter(mockContext); } private void enableReporting(boolean enabled) { - when(profile.getBoolean(DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, false)) + when(mockProfile.getBoolean(DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, true)) .thenReturn(enabled); } + /** A reporter over the bare mock context whose report is fixed (or fails) as given. */ + private DefaultDriverConfigReporter reporterReporting(Supplier json) { + return new DefaultDriverConfigReporter(mockContext) { + @Override + String buildJson(boolean scyllaDb) { + return json.get(); + } + }; + } + + // ==================== Gating ==================== + // + // Note that SESSION_ID is not this class's concern: it is an innate startup option built by + // StartupOptionsBuilder and sent on every connection regardless of these settings. + + @Test + public void should_add_driver_config_when_enabled() { + enableReporting(true); + Map options = new HashMap<>(); + reporterReporting(() -> "{\"version\":1}").populateControlConnectionOptions(options, false); + assertThat(options) + .hasSize(1) + .containsEntry(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY, "{\"version\":1}"); + } + @Test public void should_add_nothing_when_disabled() { enableReporting(false); Map options = new HashMap<>(); - reporter.populateStartupOptions(options, /* reportDriverConfig= */ true); - assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.SESSION_ID_KEY); - assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + reporter.populateControlConnectionOptions(options, false); + assertThat(options).isEmpty(); } @Test - public void should_add_session_id_and_driver_config_on_control_connection() { - enableReporting(true); + public void should_add_driver_config_when_the_option_is_not_defined() { + // A configuration that omits the option altogether must behave like the shipped default, which + // is enabled. Uses a real (map-based) profile: a mock would return false for any unstubbed + // getBoolean(), ignoring the fallback that is under test here. Map options = new HashMap<>(); - reporter.populateStartupOptions(options, /* reportDriverConfig= */ true); - // SESSION_ID is a valid, driver-generated UUID. - String sessionId = options.get(DefaultDriverConfigReporter.SESSION_ID_KEY); - assertThat(sessionId).isNotNull(); - assertThat(UUID.fromString(sessionId)).isNotNull(); // does not throw => valid UUID - // Stage 1 emits only the schema version; the value must be valid compact JSON. - assertThat(options.get(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY)) - .isEqualTo("{\"version\":" + DefaultDriverConfigReporter.SCHEMA_VERSION + "}"); + defaultsReporter(map -> map.remove(TypedDriverOption.DRIVER_CONFIG_REPORTING_ENABLED)) + .populateControlConnectionOptions(options, false); + assertThat(options).containsKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); } + // ==================== Fail-safe ==================== + @Test - public void should_add_session_id_only_on_pool_connection() { - enableReporting(true); + public void should_not_throw_when_reading_the_flag_fails() { + when(mockProfile.getBoolean(DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, true)) + .thenThrow(new IllegalStateException("config blew up")); Map options = new HashMap<>(); - reporter.populateStartupOptions(options, /* reportDriverConfig= */ false); - assertThat(options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); - assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + reporter.populateControlConnectionOptions(options, false); // must not throw + assertThat(options).isEmpty(); } @Test - public void should_use_a_stable_session_id_across_connections() { + public void should_skip_driver_config_when_building_fails() { enableReporting(true); - Map control = new HashMap<>(); - Map pool = new HashMap<>(); - reporter.populateStartupOptions(control, true); - reporter.populateStartupOptions(pool, false); - assertThat(pool.get(DefaultDriverConfigReporter.SESSION_ID_KEY)) - .isEqualTo(control.get(DefaultDriverConfigReporter.SESSION_ID_KEY)); + Map options = new HashMap<>(); + reporterReporting( + () -> { + throw new IllegalStateException("introspection blew up"); + }) + .populateControlConnectionOptions(options, false); // must not throw + assertThat(options).isEmpty(); } @Test - public void should_use_a_distinct_session_id_per_reporter() { + public void should_skip_driver_config_when_serialization_fails() { + // buildJson() returns null when Jackson fails to serialize the node tree. enableReporting(true); - Map first = new HashMap<>(); - reporter.populateStartupOptions(first, false); - // A second session (new reporter instance) must get a different SESSION_ID. - Map second = new HashMap<>(); - new DefaultDriverConfigReporter(context).populateStartupOptions(second, false); - assertThat(second.get(DefaultDriverConfigReporter.SESSION_ID_KEY)) - .isNotEqualTo(first.get(DefaultDriverConfigReporter.SESSION_ID_KEY)); + Map options = new HashMap<>(); + reporterReporting(() -> null).populateControlConnectionOptions(options, false); + assertThat(options).isEmpty(); } - /** Reporting must never break the connection: a failed config read is swallowed entirely. */ @Test - public void should_not_throw_when_reading_the_flag_fails() { - when(profile.getBoolean(DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, false)) - .thenThrow(new IllegalStateException("config blew up")); + public void should_not_throw_when_a_getSimpleName_call_throws_an_error() { + // customPolicy() calls getClass().getSimpleName() on arbitrary user-supplied policy objects; + // the catch clause must also cover Error (not just RuntimeException) so a JDK edge case there + // can never break connection setup. + enableReporting(true); Map options = new HashMap<>(); - reporter.populateStartupOptions(options, true); // must not throw - assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.SESSION_ID_KEY); - assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + reporterReporting( + () -> { + throw new InternalError("simulated getSimpleName() JDK edge case"); + }) + .populateControlConnectionOptions(options, false); // must not throw + assertThat(options).isEmpty(); } - /** - * Reporting must never break the connection: a failure while building the config groups (as a - * Stage 2 policy introspection might) is swallowed. SESSION_ID is still emitted (it is added - * before, and independently of, the DRIVER_CONFIG blob); only DRIVER_CONFIG is omitted. - */ + // ==================== Report content ==================== + @Test - public void should_keep_session_id_but_skip_driver_config_when_building_config_groups_fails() { - enableReporting(true); - DefaultDriverConfigReporter throwingReporter = - new DefaultDriverConfigReporter(context) { - @Override - protected void populateConfig(ObjectNode root, DriverExecutionProfile config) { - throw new IllegalStateException("policy introspection blew up"); - } - }; - Map options = new HashMap<>(); - throwingReporter.populateStartupOptions(options, true); // must not throw - assertThat(options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); - assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + public void should_report_default_configuration() throws Exception { + JsonNode report = report(defaultsReporter(map -> {})); + + assertThat(report.get("version").asInt()).isEqualTo(DefaultDriverConfigReporter.SCHEMA_VERSION); + + // Groups always present for the default profile. + for (String group : + new String[] { + "connection", + "socket", + "control-plane", + "reconnection-policy", + "retry-policy", + "load-balancing-policy", + "node-location-preference", + "connection-pool", + "query-defaults", + "tls" + }) { + assertThat(report.has(group)).as("group %s present", group).isTrue(); + } + // No speculative execution policy configured by default: the group has no null variant in + // the schema, so it is omitted entirely rather than reported as null. + assertThat(report.has("speculative-execution-policy")).isFalse(); + + JsonNode connection = report.get("connection"); + assertThat(connection.get("connect").get("timeout-ms").asLong()).isPositive(); + // No socket-level read/write timeout, and connection.heartbeat has no schema slot yet: all + // three are omitted rather than present-with-null/empty. + assertThat(connection.has("read")).isFalse(); + assertThat(connection.has("write")).isFalse(); + assertThat(connection.has("heartbeat")).isFalse(); + + JsonNode socket = report.get("socket"); + assertThat(socket.get("tcp-no-delay").asBoolean()).isTrue(); + assertThat(socket.get("keep-alive").asBoolean()).isFalse(); + assertThat(socket.has("linger")).isFalse(); + assertThat(socket.has("receive-buffer")).isFalse(); + assertThat(socket.has("send-buffer")).isFalse(); + + // Built without the ScyllaDB signal (plain Cassandra): no server-side (USING TIMEOUT) + // internal-query timeout, since Cassandra never gets that clause. + JsonNode controlPlane = report.get("control-plane"); + assertThat(controlPlane.get("system-queries").get("timeout").get("client-side-ms").asLong()) + .isPositive(); + assertThat(controlPlane.get("system-queries").get("timeout").has("server-side-ms")).isFalse(); + assertThat(controlPlane.get("schema-agreement").get("timeout-ms").asLong()).isPositive(); + + JsonNode reconnection = report.get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("exponential"); + assertThat(reconnection.get("base-ms").asLong()).isPositive(); + assertThat(reconnection.get("max-ms").asLong()).isPositive(); + // Java's built-in reconnection policies are unbounded: max-attempts is omitted. + assertThat(reconnection.has("max-attempts")).isFalse(); + + JsonNode retry = report.get("retry-policy"); + assertThat(retry.get("type").asText()).isEqualTo("standard-error-aware"); + assertThat(retry.has("backoff")).isFalse(); + + JsonNode lb = report.get("load-balancing-policy"); + assertThat(lb.get("type").asText()).isEqualTo("default"); + assertThat(lb.get("token-aware").asBoolean()).isTrue(); + // Both are unconditional/on-by-default for DefaultLoadBalancingPolicy: replicas are always + // shuffled, and slow-replica avoidance (the closest available signal for "latency-awareness") + // defaults to enabled. + assertThat(lb.get("shuffle").asBoolean()).isTrue(); + assertThat(lb.get("latency-awareness").asBoolean()).isTrue(); + assertThat(lb.get("dc-failover").asBoolean()).isFalse(); + // local-dc/local-rack are no longer reported here; see node-location-preference below. + assertThat(lb.has("local-dc")).isFalse(); + assertThat(lb.has("local-rack")).isFalse(); + + // local-datacenter not configured in the defaults => DC is inferred (dc-auto), and the value + // isn't known yet at report time. + JsonNode nodeLocation = report.get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc-auto"); + assertThat(nodeLocation.has("local-dc")).isFalse(); + + JsonNode pool = report.get("connection-pool"); + assertThat(pool.get("type").asText()).isEqualTo("host"); + assertThat(pool.get("desired-connections-count").asInt()).isPositive(); + assertThat(pool.get("shard-aware").get("enabled").asBoolean()).isTrue(); + + JsonNode query = report.get("query-defaults"); + assertThat(query.get("consistency").asText()).isEqualTo("LOCAL_ONE"); + assertThat(query.get("idempotence").asBoolean()).isFalse(); + assertThat(query.get("client-timestamps").asBoolean()).isTrue(); + assertThat(query.get("request").get("timeout-ms").asLong()).isPositive(); + assertThat(query.get("page").get("size").asInt()).isPositive(); + + JsonNode tls = report.get("tls"); + assertThat(tls.get("enabled").asBoolean()).isFalse(); + assertThat(tls.get("hostname-verification").asBoolean()).isFalse(); + } + + @Test + public void should_report_server_side_timeout_for_scylladb() throws Exception { + JsonNode report = report(defaultsReporter(map -> {}), /* scyllaDb= */ true); + JsonNode timeout = report.get("control-plane").get("system-queries").get("timeout"); + assertThat(timeout.get("client-side-ms").asLong()).isPositive(); + // ScyllaDB only: CassandraSchemaQueries adds a "USING TIMEOUT" clause built from this same + // option to every schema query, so it's a genuine server-side timeout on this backend. + assertThat(timeout.get("server-side-ms").asLong()).isPositive(); + } + + @Test + public void should_omit_server_side_timeout_for_cassandra() throws Exception { + JsonNode report = report(defaultsReporter(map -> {}), /* scyllaDb= */ false); + assertThat( + report.get("control-plane").get("system-queries").get("timeout").has("server-side-ms")) + .isFalse(); + } + + @Test + public void should_report_constant_reconnection_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ConstantReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode reconnection = report(r).get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("constant"); + assertThat(reconnection.get("delay-ms").asLong()).isPositive(); + assertThat(reconnection.has("max-attempts")).isFalse(); + } + + @Test + public void should_report_custom_reconnection_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ReconnectionPolicy.class), // neither exponential nor constant + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode reconnection = report(r).get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("custom"); + assertThat(reconnection.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_a_reconnection_policy_subclass_as_custom() throws Exception { + // A real (anonymous) subclass of a built-in, not a mock: proves the exact-class check doesn't + // misclassify user customizations of a built-in as the plain built-in + // (ConstantReconnectionPolicy + // is not final, and a real subclass of it already exists elsewhere in this repo's test code). + ReconnectionPolicy subclass = new ConstantReconnectionPolicy(policyConstructionContext()) {}; + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + subclass, + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode reconnection = report(r).get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("custom"); + // Also exercises the anonymous-class name fallback: getSimpleName() is empty for an anonymous + // class, so the reported name must fall back to the (non-empty) binary class name. + assertThat(reconnection.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_downgrading_consistency_retry_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(ConsistencyDowngradingRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + assertThat(report(r).get("retry-policy").get("type").asText()) + .isEqualTo("downgrading-consistency"); + } + + @Test + public void should_report_a_retry_policy_subclass_as_custom() throws Exception { + // Real (anonymous) subclass, not a mock: DefaultRetryPolicy is not final, and a real subclass + // of it already exists elsewhere in this repo's test code (osgi-tests' CustomRetryPolicy). + RetryPolicy subclass = new DefaultRetryPolicy(policyConstructionContext(), "default") {}; + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + subclass, + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode retry = report(r).get("retry-policy"); + assertThat(retry.get("type").asText()).isEqualTo("custom"); + assertThat(retry.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_constant_speculative_execution_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults( + map -> { + map.put(TypedDriverOption.SPECULATIVE_EXECUTION_MAX, 3); + map.put(TypedDriverOption.SPECULATIVE_EXECUTION_DELAY, Duration.ofMillis(100)); + }), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(ConstantSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode spec = report(r).get("speculative-execution-policy"); + assertThat(spec.get("type").asText()).isEqualTo("constant"); + assertThat(spec.get("max-executions").asInt()).isEqualTo(3); + assertThat(spec.get("delay-ms").asLong()).isEqualTo(100); + } + + @Test + public void should_report_a_speculative_execution_policy_subclass_as_custom() throws Exception { + // Real (anonymous) subclass, not a mock: NoSpeculativeExecutionPolicy is not final. A subclass + // must be reported as "custom", not silently treated the same as "no policy" (which would drop + // the whole group). + SpeculativeExecutionPolicy subclass = + new NoSpeculativeExecutionPolicy(policyConstructionContext(), "default") {}; + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + subclass, + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode spec = report(r).get("speculative-execution-policy"); + assertThat(spec.get("type").asText()).isEqualTo("custom"); + assertThat(spec.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_dc_inferring_load_balancing_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DcInferringLoadBalancingPolicy.class), // extends DefaultLoadBalancingPolicy + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode lb = report(r).get("load-balancing-policy"); + // Must be its own "dc-inferring" type, not misclassified as "default" via instanceof. + assertThat(lb.get("type").asText()).isEqualTo("dc-inferring"); + assertThat(lb.get("token-aware").asBoolean()).isTrue(); + assertThat(lb.get("shuffle").asBoolean()).isTrue(); + } + + @Test + public void should_report_dse_load_balancing_policy_as_default() throws Exception { + // DseLoadBalancingPolicy is a deprecated, behavior-identical alias of + // DefaultLoadBalancingPolicy; must not fall through to "custom". Note: a real (non-mocked) + // instance of this class requires a resolvable local DC (it uses MandatoryLocalDcHelper) and + // would fail to construct with no DC configured, unlike DcInferringLoadBalancingPolicy below; + // the mock here bypasses that constructor validation, same as + // should_report_default_configuration. + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DseLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + assertThat(report(r).get("load-balancing-policy").get("type").asText()).isEqualTo("default"); + } + + @Test + public void should_report_dse_dc_inferring_load_balancing_policy() throws Exception { + // DseDcInferringLoadBalancingPolicy is a deprecated, behavior-identical alias of + // DcInferringLoadBalancingPolicy; must not fall through to "custom". + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DseDcInferringLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + assertThat(report(r).get("load-balancing-policy").get("type").asText()) + .isEqualTo("dc-inferring"); + } + + @Test + public void should_report_basic_load_balancing_policy() throws Exception { + // A real, distinct, documented third built-in (reference.conf lists exactly three); must be + // reported as its own "basic" type, not misclassified as "custom". + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(BasicLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode lb = report(r).get("load-balancing-policy"); + assertThat(lb.get("type").asText()).isEqualTo("basic"); + assertThat(lb.get("token-aware").asBoolean()).isTrue(); + assertThat(lb.get("shuffle").asBoolean()).isTrue(); + // Unlike DefaultLoadBalancingPolicy, BasicLoadBalancingPolicy has no slow-replica-avoidance + // mechanism at all: always false, regardless of the (default-policy-only) config option. + assertThat(lb.get("latency-awareness").asBoolean()).isFalse(); + } + + @Test + public void should_report_latency_awareness_disabled_when_slow_avoidance_is_off() + throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter( + map -> map.put(TypedDriverOption.LOAD_BALANCING_POLICY_SLOW_AVOIDANCE, false)); + assertThat(report(r).get("load-balancing-policy").get("latency-awareness").asBoolean()) + .isFalse(); + } + + @Test + public void should_report_custom_load_balancing_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(LoadBalancingPolicy.class), // not the default policy + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode lb = report(r).get("load-balancing-policy"); + assertThat(lb.get("type").asText()).isEqualTo("custom"); + assertThat(lb.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_explicit_local_dc() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc1")); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc1"); + assertThat(nodeLocation.has("local-rack")).isFalse(); + } + + @Test + public void should_report_explicit_local_dc_and_rack() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter( + map -> { + map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc1"); + map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_RACK, "rack1"); + }); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("rack"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc1"); + assertThat(nodeLocation.get("local-rack").asText()).isEqualTo("rack1"); + } + + @Test + public void should_report_local_dc_set_via_session_builder() throws Exception { + // SessionBuilder.withLocalDatacenter(...), not the config option: surfaced through + // InternalDriverContext.getLocalDatacenter(), which the reporter must consult. + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty(), + /* programmaticLocalDc= */ "dc-programmatic"); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc-programmatic"); + } + + @Test + public void should_prefer_programmatic_local_dc_over_config_option() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults( + map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc-config")), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty(), + /* programmaticLocalDc= */ "dc-programmatic"); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc-programmatic"); + } + + @Test + public void should_report_rack_auto_when_only_rack_is_configured() throws Exception { + // Rack configured explicitly, but no DC (neither programmatically nor via config): the DC will + // be inferred, so this must not silently drop the explicitly-configured rack. + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_RACK, "rack1")); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("rack-auto"); + assertThat(nodeLocation.get("local-rack").asText()).isEqualTo("rack1"); + assertThat(nodeLocation.has("local-dc")).isFalse(); + } + + @Test + public void should_report_server_side_timestamps_as_disabled_client_timestamps() + throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(ServerSideTimestampGenerator.class), + Optional.empty()); + assertThat(report(r).get("query-defaults").get("client-timestamps").asBoolean()).isFalse(); + } + + @Test + public void should_report_tls_enabled_with_hostname_verification() throws Exception { + // hostname-verification comes from the factory's own accessor, not the config option. + SslEngineFactory factory = mock(SslEngineFactory.class); + when(factory.requireHostnameValidation()).thenReturn(true); + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.of(factory)); + JsonNode tls = report(r).get("tls"); + assertThat(tls.get("enabled").asBoolean()).isTrue(); + assertThat(tls.get("hostname-verification").asBoolean()).isTrue(); + } + + @Test + public void should_report_hostname_verification_from_factory_not_config_option() + throws Exception { + // Regression for the false-report bug: a ProgrammaticSslEngineFactory (as built by + // SessionBuilder.withSslContext(...)) does NO hostname validation by default and ignores the + // SSL_HOSTNAME_VALIDATION config option. The report must reflect the factory's real state + // (false), not the config option (true here) — otherwise it falsely claims validation is on. + SslEngineFactory programmatic = new ProgrammaticSslEngineFactory(SSLContext.getDefault()); + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> map.put(TypedDriverOption.SSL_HOSTNAME_VALIDATION, true)), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.of(programmatic)); + JsonNode tls = report(r).get("tls"); + assertThat(tls.get("enabled").asBoolean()).isTrue(); + assertThat(tls.get("hostname-verification").asBoolean()).isFalse(); + } + + @Test + public void should_report_socket_overrides() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter( + map -> { + map.put(TypedDriverOption.SOCKET_KEEP_ALIVE, true); + map.put(TypedDriverOption.SOCKET_RECEIVE_BUFFER_SIZE, 65535); + map.put(TypedDriverOption.SOCKET_LINGER_INTERVAL, 5); + }); + JsonNode socket = report(r).get("socket"); + assertThat(socket.get("keep-alive").asBoolean()).isTrue(); + assertThat(socket.get("receive-buffer").get("size-bytes").asInt()).isEqualTo(65535); + assertThat(socket.get("linger").get("interval-s").asInt()).isEqualTo(5); + } + + @Test + public void should_omit_page_when_unbounded() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.REQUEST_PAGE_SIZE, 0)); + // The schema has no "unbounded" sentinel: the whole page group is omitted instead. + assertThat(report(r).get("query-defaults").has("page")).isFalse(); + } + + @Test + public void should_report_bounded_page_size() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.REQUEST_PAGE_SIZE, 5000)); + assertThat(report(r).get("query-defaults").get("page").get("size").asInt()).isEqualTo(5000); + } + + // ==================== Schema conformance ==================== + // + // These build a config, serialize it via the reporter, and validate the produced JSON against the + // normative v1 JSON Schema (the same document ScyllaDB uses to interpret DRIVER_CONFIG). They + // cover every discriminated-union branch and optional-group case the reporter can emit, turning + // the "every emitted document is schema-valid" invariant into an enforced test. + + @Test + public void should_conform_to_schema_for_default_report() throws Exception { + assertConformsToSchema(report(defaultsReporter(map -> {}))); + } + + @Test + public void should_conform_to_schema_for_default_report_on_scylladb() throws Exception { + // scyllaDb=true adds control-plane.system-queries.timeout.server-side-ms. + assertConformsToSchema(report(defaultsReporter(map -> {}), /* scyllaDb= */ true)); + } + + @Test + public void should_conform_to_schema_for_constant_reconnection_policy() throws Exception { + assertConformsToSchema( + report( + reporterWith( + defaults(map -> {}), + mock(ConstantReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()))); + } + + @Test + public void should_conform_to_schema_for_custom_reconnection_policy() throws Exception { + assertConformsToSchema( + report( + reporterWith( + defaults(map -> {}), + mock(ReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()))); + } + + @Test + public void should_conform_to_schema_for_downgrading_consistency_retry_policy() throws Exception { + assertConformsToSchema( + report( + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(ConsistencyDowngradingRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()))); + } + + @Test + public void should_conform_to_schema_for_custom_retry_policy() throws Exception { + assertConformsToSchema( + report( + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(RetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()))); + } + + @Test + public void should_conform_to_schema_for_constant_speculative_execution_policy() + throws Exception { + assertConformsToSchema( + report( + reporterWith( + defaults( + map -> { + map.put(TypedDriverOption.SPECULATIVE_EXECUTION_MAX, 3); + map.put( + TypedDriverOption.SPECULATIVE_EXECUTION_DELAY, Duration.ofMillis(100)); + }), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(ConstantSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()))); + } + + @Test + public void should_conform_to_schema_for_basic_load_balancing_policy() throws Exception { + assertConformsToSchema( + report( + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(BasicLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()))); + } + + @Test + public void should_conform_to_schema_for_custom_load_balancing_policy() throws Exception { + assertConformsToSchema( + report( + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(LoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()))); + } + + @Test + public void should_conform_to_schema_for_explicit_dc_and_rack() throws Exception { + assertConformsToSchema( + report( + defaultsReporter( + map -> { + map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc1"); + map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_RACK, "rack1"); + }))); + } + + @Test + public void should_conform_to_schema_for_rack_auto_node_location() throws Exception { + assertConformsToSchema( + report( + defaultsReporter( + map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_RACK, "rack1")))); + } + + @Test + public void should_conform_to_schema_for_tls_enabled_with_hostname_verification() + throws Exception { + SslEngineFactory factory = mock(SslEngineFactory.class); + when(factory.requireHostnameValidation()).thenReturn(true); + assertConformsToSchema( + report( + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.of(factory)))); + } + + @Test + public void should_conform_to_schema_for_socket_overrides() throws Exception { + assertConformsToSchema( + report( + defaultsReporter( + map -> { + map.put(TypedDriverOption.SOCKET_KEEP_ALIVE, true); + map.put(TypedDriverOption.SOCKET_RECEIVE_BUFFER_SIZE, 65535); + map.put(TypedDriverOption.SOCKET_SEND_BUFFER_SIZE, 65535); + map.put(TypedDriverOption.SOCKET_LINGER_INTERVAL, 5); + }))); + } + + @Test + public void should_reject_a_report_that_violates_the_schema() throws Exception { + // Sanity check that the validator actually enforces the schema (rather than accepting + // anything): + // an unknown top-level key must be rejected, since the schema sets additionalProperties=false. + ObjectNode report = (ObjectNode) report(defaultsReporter(map -> {})); + report.put("bogus-unknown-key", "x"); + assertThat(SCHEMA.validate(report)).as("unknown top-level key must be rejected").isNotEmpty(); + } + + // ==================== helpers ==================== + + private void assertConformsToSchema(JsonNode report) { + Set errors = SCHEMA.validate(report); + assertThat(errors).as("schema violations in %s", report).isEmpty(); + } + + private JsonNode report(DefaultDriverConfigReporter reporter) throws Exception { + return report(reporter, /* scyllaDb= */ false); + } + + private JsonNode report(DefaultDriverConfigReporter reporter, boolean scyllaDb) throws Exception { + return MAPPER.readTree(reporter.buildJson(scyllaDb)); + } + + /** A real default execution profile with the given customizations applied. */ + private DriverExecutionProfile defaults(Consumer customizer) { + OptionsMap map = OptionsMap.driverDefaults(); + customizer.accept(map); + return DriverConfigLoader.fromMap(map).getInitialConfig().getDefaultProfile(); + } + + /** Reporter over default config + the Java-default policy set. */ + private DefaultDriverConfigReporter defaultsReporter(Consumer customizer) { + return reporterWith( + defaults(customizer), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + } + + private DefaultDriverConfigReporter reporterWith( + DriverExecutionProfile profile, + ReconnectionPolicy reconnection, + RetryPolicy retry, + SpeculativeExecutionPolicy speculative, + LoadBalancingPolicy loadBalancing, + TimestampGenerator timestamps, + Optional ssl) { + return reporterWith( + profile, reconnection, retry, speculative, loadBalancing, timestamps, ssl, null); + } + + /** Same as the 7-arg overload, with an optional programmatic ({@code withLocalDatacenter}) DC. */ + private DefaultDriverConfigReporter reporterWith( + DriverExecutionProfile profile, + ReconnectionPolicy reconnection, + RetryPolicy retry, + SpeculativeExecutionPolicy speculative, + LoadBalancingPolicy loadBalancing, + TimestampGenerator timestamps, + Optional ssl, + String programmaticLocalDc) { + InternalDriverContext ctx = mock(InternalDriverContext.class); + DriverConfig config = mock(DriverConfig.class); + when(ctx.getConfig()).thenReturn(config); + when(config.getDefaultProfile()).thenReturn(profile); + when(ctx.getReconnectionPolicy()).thenReturn(reconnection); + when(ctx.getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME)).thenReturn(retry); + when(ctx.getSpeculativeExecutionPolicy(DriverExecutionProfile.DEFAULT_NAME)) + .thenReturn(speculative); + when(ctx.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME)).thenReturn(loadBalancing); + when(ctx.getTimestampGenerator()).thenReturn(timestamps); + when(ctx.getSslEngineFactory()).thenReturn(ssl); + when(ctx.getLocalDatacenter(DriverExecutionProfile.DEFAULT_NAME)) + .thenReturn(programmaticLocalDc); + return new DefaultDriverConfigReporter(ctx); + } + + /** A minimal {@link DriverContext} good enough to construct a real built-in policy instance. */ + private DriverContext policyConstructionContext() { + DriverContext ctx = mock(DriverContext.class); + DriverConfig config = mock(DriverConfig.class); + DriverExecutionProfile profile = defaults(map -> {}); + when(ctx.getConfig()).thenReturn(config); + when(config.getDefaultProfile()).thenReturn(profile); + when(config.getProfile(DriverExecutionProfile.DEFAULT_NAME)).thenReturn(profile); + when(ctx.getSessionName()).thenReturn("test-session"); + return ctx; } } 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 2f8f4174093..963e9954592 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 @@ -30,6 +30,7 @@ import com.tngtech.java.junit.dataprovider.DataProvider; import com.tngtech.java.junit.dataprovider.DataProviderRunner; import java.util.Optional; +import java.util.UUID; import org.junit.Test; import org.junit.runner.RunWith; @@ -54,6 +55,10 @@ private void assertDefaultStartupOptions(Startup startup) { assertThat(startup.options).containsKey(StartupOptionsBuilder.DRIVER_VERSION_KEY); Version version = Version.parse(startup.options.get(StartupOptionsBuilder.DRIVER_VERSION_KEY)); assertThat(version).isEqualByComparingTo(Session.OSS_DRIVER_COORDINATES.getVersion()); + // SESSION_ID is innate: sent on every connection, whatever the configuration says. + assertThat(startup.options).containsKey(StartupOptionsBuilder.SESSION_ID_KEY); + assertThat(UUID.fromString(startup.options.get(StartupOptionsBuilder.SESSION_ID_KEY))) + .isNotNull(); } @Test @@ -85,6 +90,35 @@ public void should_build_startup_options(String compression) { assertDefaultStartupOptions(startup); } + @Test + public void should_use_a_stable_session_id_for_the_whole_session() { + + // The startup options are built once per session and copied into every connection's STARTUP, so + // all of a session's connections report the same SESSION_ID. + DefaultDriverContext ctx = MockedDriverContextFactory.defaultDriverContext(); + assertThat(ctx.getStartupOptions().get(StartupOptionsBuilder.SESSION_ID_KEY)) + .isEqualTo(ctx.getStartupOptions().get(StartupOptionsBuilder.SESSION_ID_KEY)); + } + + @Test + public void should_use_a_distinct_session_id_per_session() { + + DefaultDriverContext ctx1 = MockedDriverContextFactory.defaultDriverContext(); + DefaultDriverContext ctx2 = MockedDriverContextFactory.defaultDriverContext(); + assertThat(ctx1.getStartupOptions().get(StartupOptionsBuilder.SESSION_ID_KEY)) + .isNotEqualTo(ctx2.getStartupOptions().get(StartupOptionsBuilder.SESSION_ID_KEY)); + } + + @Test + public void should_not_derive_session_id_from_client_id() { + + // SESSION_ID must be driver-generated, not the (user-settable) CLIENT_ID, so that it is + // guaranteed unique per session as the grouping key requires. + DefaultDriverContext ctx = MockedDriverContextFactory.defaultDriverContext(); + assertThat(ctx.getStartupOptions().get(StartupOptionsBuilder.SESSION_ID_KEY)) + .isNotEqualTo(ctx.getStartupOptions().get(StartupOptionsBuilder.CLIENT_ID_KEY)); + } + @Test public void should_fail_to_build_startup_options_with_invalid_compression() { diff --git a/core/src/test/resources/config/driver-config-report-v1.schema.json b/core/src/test/resources/config/driver-config-report-v1.schema.json new file mode 100644 index 00000000000..b2b5886bb7c --- /dev/null +++ b/core/src/test/resources/config/driver-config-report-v1.schema.json @@ -0,0 +1,846 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://scylladb.com/schemas/driver-client-options/v1.json", + "title": "ScyllaDB driver DRIVER_CONFIG configuration", + "description": "Schema for the JSON value sent under the STARTUP option key DRIVER_CONFIG, describing the effective client configuration. The top-level object must include `version` and the required configuration groups listed by this schema. Unknown top-level keys are rejected. Built-in groups reject unknown keys and require the keys listed in each group; custom policy objects may include additional implementation-specific public attributes where explicitly allowed.", + "type": "object", + "additionalProperties": false, + "required": [ + "version", + "connection", + "socket", + "control-plane", + "reconnection-policy", + "retry-policy", + "load-balancing-policy", + "connection-pool", + "query-defaults", + "tls" + ], + "properties": { + "version": { + "description": "Major schema version. Adding keys is backward-compatible and does not bump this; only changing/removing the meaning of an existing key does.", + "type": "integer", + "const": 1 + }, + "connection": { + "$ref": "#/$defs/connection" + }, + "socket": { + "$ref": "#/$defs/socket" + }, + "control-plane": { + "$ref": "#/$defs/control-plane" + }, + "reconnection-policy": { + "$ref": "#/$defs/reconnection-policy" + }, + "retry-policy": { + "$ref": "#/$defs/retry-policy" + }, + "speculative-execution-policy": { + "$ref": "#/$defs/speculative-execution-policy" + }, + "load-balancing-policy": { + "$ref": "#/$defs/load-balancing-policy" + }, + "node-location-preference": { + "$ref": "#/$defs/node-location-preference" + }, + "connection-pool": { + "$ref": "#/$defs/connection-pool" + }, + "query-defaults": { + "$ref": "#/$defs/query-defaults" + }, + "tls": { + "$ref": "#/$defs/tls" + } + }, + "$defs": { + "positiveInteger": { + "type": "integer", + "minimum": 1 + }, + "nonNegativeInteger": { + "type": "integer", + "minimum": 0 + }, + "retryPolicyBackoff": { + "description": "Delay inserted between retry attempts of a retry policy. Discriminated union: when present, `type` selects the backoff algorithm and each algorithm carries only its own parameters. Absent when there is no delay between attempts.", + "oneOf": [ + { + "type": "object", + "description": "Exponential backoff: the delay starts at base-ms and doubles after each attempt (capped at max-ms), with a small random jitter to de-synchronize concurrent retries.", + "additionalProperties": false, + "required": [ + "type", + "base-ms" + ], + "properties": { + "type": { + "const": "exponential", + "description": "Exponential backoff algorithm." + }, + "base-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Initial delay between retries in milliseconds; the starting delay that doubles each attempt." + }, + "max-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum delay between retries in milliseconds; the exponentially growing delay is capped here. Absent when no maximum delay is configured." + } + } + }, + { + "type": "object", + "description": "Constant backoff: a fixed delay is inserted between every retry attempt.", + "additionalProperties": false, + "required": [ + "type", + "delay-ms" + ], + "properties": { + "type": { + "const": "constant", + "description": "Constant (fixed-delay) backoff algorithm." + }, + "delay-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Fixed delay between retries in milliseconds." + } + } + } + ] + }, + "connection": { + "description": "Connection-level settings: socket read/write/connect timeouts plus the CQL-level idle heartbeat. Durations are in milliseconds. Optional duration fields are absent when unset or not applicable.", + "type": "object", + "required": [ + "connect" + ], + "additionalProperties": false, + "properties": { + "connect": { + "type": "object", + "description": "Settings for establishing a TCP/CQL connection to a node.", + "required": [ + "timeout-ms" + ], + "additionalProperties": false, + "properties": { + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Timeout for establishing a TCP/CQL connection to a node." + } + } + }, + "read": { + "type": "object", + "description": "Settings for reading from a connection.", + "additionalProperties": false, + "properties": { + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Read operation timeout." + } + } + }, + "write": { + "type": "object", + "description": "Settings for writing to a connection. Direction-specific options such as write coalescing are expected to be added here in a future schema version.", + "additionalProperties": false, + "properties": { + "coalescing": { + "type": "object", + "description": "Settings for write coalescing. It is a placeholder for v2", + "additionalProperties": false, + "properties": {} + }, + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Write operation timeout." + } + } + }, + "heartbeat": { + "type": "object", + "description": "Reserved for CQL-level idle heartbeat settings. Optional and intentionally empty in this schema version. It is a placeholder for v2", + "additionalProperties": false, + "properties": {} + } + } + }, + "control-plane": { + "description": "Control-plane timeout settings for internal/system queries run over the control connection and for schema agreement. Each value is in milliseconds. Optional values are absent when unset or not applicable.", + "type": "object", + "required": [ + "system-queries", + "schema-agreement" + ], + "additionalProperties": false, + "properties": { + "system-queries": { + "type": "object", + "description": "Settings for internal/system queries run over the control connection.", + "additionalProperties": false, + "required": [ + "timeout" + ], + "properties": { + "timeout": { + "type": "object", + "description": "Timeouts applied to internal/system queries. Each value is in milliseconds. Optional values are absent when unset or not applicable.", + "additionalProperties": false, + "properties": { + "client-side-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "A client-side timeout for internal queries." + }, + "server-side-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "A server-side timeout for internal queries." + } + } + } + } + }, + "schema-agreement": { + "type": "object", + "description": "Settings for schema agreement across nodes.", + "additionalProperties": false, + "required": [ + "timeout-ms" + ], + "properties": { + "timeout-ms": { + "$ref": "#/$defs/nonNegativeInteger", + "description": "Maximum time to wait for schema agreement across nodes. Always a concrete value; 0 means do not wait for agreement." + } + } + } + } + }, + "socket": { + "description": "Low-level TCP socket options applied to client connections. Boolean options (tcp-no-delay, keep-alive, reuse-address) report the effective on/off state: when no explicit value is configured, the OS/platform default is reported. Buffer sizes are in bytes and linger is in seconds; these fields are absent when unset (kernel auto-tuned buffer / linger disabled).", + "type": "object", + "required": [ + "tcp-no-delay", + "keep-alive", + "reuse-address" + ], + "additionalProperties": false, + "properties": { + "tcp-no-delay": { + "type": "boolean", + "description": "TCP_NODELAY: disable Nagle's algorithm. Reports the effective value; when no explicit value is configured, the OS/platform default is reported." + }, + "keep-alive": { + "type": "boolean", + "description": "SO_KEEPALIVE: OS-level TCP keep-alive probes on idle connections. Reports the effective on/off state; when no explicit value is configured, the OS/platform default is reported." + }, + "reuse-address": { + "type": "boolean", + "description": "SO_REUSEADDR: allow reuse of a local address. Reports the effective on/off state; when no explicit value is configured, the OS/platform default is reported." + }, + "linger": { + "type": "object", + "required": [ + "interval-s" + ], + "additionalProperties": false, + "properties": { + "interval-s": { + "$ref": "#/$defs/nonNegativeInteger", + "description": "SO_LINGER lingering-close interval in seconds." + } + } + }, + "receive-buffer": { + "type": "object", + "required": [ + "size-bytes" + ], + "additionalProperties": false, + "properties": { + "size-bytes": { + "$ref": "#/$defs/positiveInteger", + "description": "SO_RCVBUF socket receive buffer size hint in bytes." + } + } + }, + "send-buffer": { + "type": "object", + "required": [ + "size-bytes" + ], + "additionalProperties": false, + "properties": { + "size-bytes": { + "$ref": "#/$defs/positiveInteger", + "description": "SO_SNDBUF socket send buffer size hint in bytes." + } + } + } + } + }, + "reconnection-policy": { + "description": "Defines how connection attempts to a node are retried after a connection failure.", + "oneOf": [ + { + "type": "object", + "description": "Exponential backoff reconnection policy.", + "additionalProperties": false, + "required": [ + "type", + "base-ms", + "max-ms" + ], + "properties": { + "type": { + "const": "exponential", + "description": "Reconnection policy type." + }, + "base-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Initial delay before the first reconnection attempt in milliseconds. Always a concrete value when this policy is reported." + }, + "max-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum delay between reconnection attempts (in milliseconds). Always a concrete value when this policy is reported." + }, + "max-attempts": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of reconnection attempts before giving up. Absent when attempts are unlimited." + } + } + }, + { + "type": "object", + "description": "Constant delay reconnection policy.", + "additionalProperties": false, + "required": [ + "type", + "delay-ms" + ], + "properties": { + "type": { + "const": "constant", + "description": "Reconnection policy type." + }, + "delay-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Fixed delay between reconnection attempts (in milliseconds). Always a concrete value when this policy is reported." + }, + "max-attempts": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of reconnection attempts before giving up. Absent when attempts are unlimited." + } + } + }, + { + "type": "object", + "description": "A user-supplied reconnection policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Reconnection policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + } + } + }, + { + "type": "null", + "description": "No reconnection attempts will be made." + } + ] + }, + "retry-policy": { + "description": "Controls whether and how a failed query is retried. Discriminated on `type`; each policy only permits its own parameters.", + "oneOf": [ + { + "type": "object", + "description": "Error-type-aware retry policy with fixed, non-configurable rules: retry at most once on Unavailable (on the next node), retry on a read timeout only when enough replicas responded but the data was not retrieved, retry on a write timeout only for batch-log writes, retry the next node on Overloaded/ServerError/Bootstrapping/broken-connection when the request is idempotent, and never retry serial (LWT) reads. It can have backoff enabled.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "standard-error-aware", + "description": "Retry policy type." + }, + "backoff": { + "$ref": "#/$defs/retryPolicyBackoff", + "description": "Delay inserted between retries." + } + } + }, + { + "type": "object", + "description": "Simple retry policy with a fixed number of retries.", + "additionalProperties": false, + "required": [ + "type", + "max-retries" + ], + "properties": { + "type": { + "const": "simple", + "description": "Retry policy type." + }, + "backoff": { + "$ref": "#/$defs/retryPolicyBackoff", + "description": "Delay inserted between retries." + }, + "max-retries": { + "$ref": "#/$defs/nonNegativeInteger", + "description": "Maximum number of retries before giving up. Always a concrete value when this policy is reported; 0 means no retries." + } + } + }, + { + "type": "object", + "description": "Fall-through retry policy: never retries anything and always rethrows the original error to the caller. Every error type — read timeout, write timeout, unavailable, and unexpected request errors (connection errors, Overloaded, ServerError, Bootstrapping) — is propagated unchanged. This is a true no-op and is stricter than the 'never' policy, which still retries the next host on connection/server errors.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "fallthrough", + "description": "Retry policy type." + } + } + }, + { + "type": "object", + "description": "Downgrading-consistency retry policy: retries at a lower consistency level on failure.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "downgrading-consistency", + "description": "Retry policy type." + }, + "backoff": { + "$ref": "#/$defs/retryPolicyBackoff", + "description": "Delay inserted between retries." + } + } + }, + { + "type": "object", + "description": "A user-supplied retry policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Retry policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + }, + "description": { + "type": "string", + "description": "Textual description of what this policy does." + } + } + } + ] + }, + "speculative-execution-policy": { + "description": "Controls pre-emptive duplicate requests to other replicas. Discriminated on `type`; each policy only permits its own parameters.", + "oneOf": [ + { + "type": "object", + "description": "Constant-delay speculative execution: launch extra executions after a fixed delay.", + "additionalProperties": false, + "required": [ + "type", + "max-executions", + "delay-ms" + ], + "properties": { + "type": { + "const": "constant", + "description": "Speculative execution policy type." + }, + "max-executions": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of speculative executions per request." + }, + "delay-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Delay before launching each additional execution (in milliseconds)." + } + } + }, + { + "type": "object", + "description": "Percentile-based speculative execution: launch extra executions once latency exceeds a percentile threshold.", + "additionalProperties": false, + "required": [ + "type", + "max-executions", + "percentile" + ], + "properties": { + "type": { + "const": "percentile", + "description": "Speculative execution policy type." + }, + "max-executions": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of speculative executions per request." + }, + "percentile": { + "type": "number", + "exclusiveMinimum": 0, + "exclusiveMaximum": 100, + "description": "Latency percentile (0–100, exclusive; e.g. 99.0) that triggers an additional execution." + } + } + }, + { + "type": "object", + "description": "A user-supplied speculative execution policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Speculative execution policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + }, + "description": { + "type": "string", + "description": "Textual description of what this policy does." + } + } + } + ] + }, + "load-balancing-policy": { + "description": "Load balancing / host selection policy. Discriminated on `type`: a built-in policy reports the normalized flags below, while a user-supplied policy is reported as type 'custom' with a `name` and, optionally, serialized public attributes.", + "oneOf": [ + { + "type": "object", + "description": "A built-in load balancing policy, reported with normalized location/awareness flags.", + "additionalProperties": false, + "required": [ + "type", + "shuffle", + "token-aware", + "dc-failover", + "latency-awareness" + ], + "properties": { + "type": { + "enum": [ + "token-aware", + "round-robin", + "dc-aware", + "rack-aware", + "dc-inferring", + "basic", + "white-list", + "host-filter", + "default" + ], + "description": "Policy type one of e.g. token-aware, round-robin, dc-aware, rack-aware." + }, + "token-aware": { + "type": "boolean", + "description": "Whether queries are routed to token replicas." + }, + "shuffle": { + "type": "boolean", + "description": "Whether candidate order in a query plan is shuffled for regular queries. LWT and strongly consistent queries are not shuffled." + }, + "dc-failover": { + "type": "boolean", + "description": "Whether requests may fail over to remote datacenters." + }, + "latency-awareness": { + "type": "boolean", + "description": "Whether latency-aware host ordering is enabled." + } + } + }, + { + "type": "object", + "description": "A user-supplied load balancing policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Load balancing policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + }, + "description": { + "type": "string", + "description": "Textual description of what this policy does." + } + } + } + ] + }, + "node-location-preference": { + "description": "Session-level datacenter/rack preference, set independently of the load balancing policy. Some implementations let users set a preferred DC/rack directly on the session configuration; the load balancing policy and other components read this preference unless a policy overrides it. May be sourced from different places; if DC/rack preferences are specified in the load balancing policy, they should be reported here.", + "oneOf": [ + { + "type": "object", + "description": "Explicitly configured datacenter preference.", + "additionalProperties": false, + "required": [ + "type", + "local-dc" + ], + "properties": { + "type": { + "const": "dc", + "description": "Session-level location preference: explicit datacenter." + }, + "local-dc": { + "type": "string", + "description": "Explicitly configured preferred datacenter." + } + } + }, + { + "type": "object", + "description": "Explicitly configured datacenter and rack preference.", + "additionalProperties": false, + "required": [ + "type", + "local-dc", + "local-rack" + ], + "properties": { + "type": { + "const": "rack", + "description": "Session-level location preference: explicit datacenter and rack." + }, + "local-dc": { + "type": "string", + "description": "Explicitly configured preferred datacenter." + }, + "local-rack": { + "type": "string", + "description": "Explicitly configured preferred rack." + } + } + }, + { + "type": "object", + "description": "Datacenter preference inferred from the first node the client connects to.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "dc-auto", + "description": "Session-level location preference: inferred datacenter." + }, + "local-dc": { + "type": "string", + "description": "Inferred preferred datacenter. Absent when not yet known at report time." + } + } + }, + { + "type": "object", + "description": "Datacenter and rack preference inferred from the first node the client connects to.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "rack-auto", + "description": "Session-level location preference: inferred datacenter and rack." + }, + "local-dc": { + "type": "string", + "description": "Inferred preferred datacenter. Absent when not yet known at report time." + }, + "local-rack": { + "type": "string", + "description": "Inferred preferred rack. Absent when not yet known at report time." + } + } + } + ] + }, + "connection-pool": { + "description": "Connection pooling configuration.", + "type": "object", + "required": [ + "type", + "desired-connections-count", + "shard-aware" + ], + "additionalProperties": false, + "properties": { + "type": { + "enum": [ + "host", + "shard" + ], + "description": "What each pool is keyed by: per host or per shard." + }, + "desired-connections-count": { + "$ref": "#/$defs/positiveInteger", + "description": "Number of connections to open per host or per shard." + }, + "connection": { + "type": "object", + "description": "Per-connection pool settings.", + "required": [ + "max-requests" + ], + "additionalProperties": false, + "properties": { + "max-requests": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of in-flight requests per connection." + } + } + }, + "shard-aware": { + "type": "object", + "required": [ + "enabled" + ], + "additionalProperties": false, + "properties": { + "enabled": { + "type": "boolean", + "description": "Whether the client is configured to use ScyllaDB's dedicated shard-aware port (default 19042, TLS 19043) to reach a chosen shard in a single connect, versus the fallback of opening connections on the normal port and reading the server-assigned shard. Reports configuration intent; at runtime the port must also be advertised by the server and reachable, otherwise the client falls back transparently." + } + } + } + } + }, + "query-defaults": { + "description": "Default per-request settings applied to statements that do not override them.", + "type": "object", + "required": [ + "consistency", + "idempotence", + "client-timestamps", + "request" + ], + "additionalProperties": false, + "properties": { + "page": { + "type": "object", + "required": [ + "size" + ], + "additionalProperties": false, + "properties": { + "size": { + "$ref": "#/$defs/positiveInteger", + "description": "Default page (fetch) size for result sets. Absent when page is not limited." + } + } + }, + "consistency": { + "description": "Default consistency level applied to requests that do not override it. Always present when this group is reported.", + "type": "string", + "enum": [ + "ANY", + "ONE", + "TWO", + "THREE", + "QUORUM", + "ALL", + "LOCAL_QUORUM", + "EACH_QUORUM", + "LOCAL_ONE" + ] + }, + "serial-consistency": { + "description": "Default serial consistency for LWT/conditional statements. Absent when unset; the server default applies.", + "type": "string", + "enum": [ + "SERIAL", + "LOCAL_SERIAL" + ] + }, + "idempotence": { + "description": "Default idempotence flag applied to statements that do not set their own.", + "type": "boolean" + }, + "client-timestamps": { + "description": "True when the client assigns the write timestamp client-side (protocol-level/USING TIMESTAMP) instead of letting the coordinator assign it.", + "type": "boolean" + }, + "request": { + "type": "object", + "description": "Default request-level settings.", + "required": [ + "timeout-ms" + ], + "additionalProperties": false, + "properties": { + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Client-side timeout for a single request/query in milliseconds. Always a concrete value when query defaults are reported." + } + } + } + } + }, + "tls": { + "description": "TLS/SSL transport settings. Reports only booleans; never credentials, keys, or host lists.", + "type": "object", + "required": [ + "enabled", + "hostname-verification" + ], + "additionalProperties": false, + "properties": { + "enabled": { + "type": "boolean", + "description": "Whether TLS is enabled for connections." + }, + "hostname-verification": { + "type": "boolean", + "description": "Whether the server hostname is verified against its certificate." + } + } + } + } +} diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingAssertions.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingAssertions.java new file mode 100644 index 00000000000..a7663b7b1f8 --- /dev/null +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingAssertions.java @@ -0,0 +1,63 @@ +/* + * 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.core.config; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.json.JsonMapper; + +/** Shared assertions for the driver-config-reporting integration tests. */ +class DriverConfigReportingAssertions { + + // FAIL_ON_TRAILING_TOKENS rejects a valid JSON value followed by garbage; the payload is read + // below via readValue(..), which honors this feature reliably (readTree historically does not). + private static final ObjectMapper OBJECT_MAPPER = + JsonMapper.builder().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS).build(); + + private DriverConfigReportingAssertions() {} + + /** + * Asserts that a {@code DRIVER_CONFIG} value is a well-formed stage-2 report: valid JSON whose + * {@code version} is the integer {@code 1} and that carries the full configuration payload + * (checked here via the always-present, backend-agnostic {@code load-balancing-policy} group). + * Guards against an incorrect schema version, a malformed blob, or an empty/stage-1-only payload + * slipping through a mere key-presence check. + */ + static void assertDriverConfigPayload(String driverConfig) { + JsonNode root; + try { + root = OBJECT_MAPPER.readValue(driverConfig, JsonNode.class); + } catch (JsonProcessingException e) { + throw new AssertionError("DRIVER_CONFIG is not valid JSON: " + driverConfig, e); + } + assertThat(root.path("version").isInt()) + .as("version is an integer in %s", driverConfig) + .isTrue(); + assertThat(root.path("version").intValue()).isEqualTo(1); + assertThat(root.path("load-balancing-policy").isObject()) + .as("load-balancing-policy is an object in %s", driverConfig) + .isTrue(); + assertThat(root.path("load-balancing-policy").path("type").isTextual()) + .as("load-balancing-policy.type is present in %s", driverConfig) + .isTrue(); + } +} diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingCcmIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingCcmIT.java index 4710f9d4efe..86ccb41f647 100644 --- a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingCcmIT.java +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingCcmIT.java @@ -17,6 +17,7 @@ */ package com.datastax.oss.driver.core.config; +import static com.datastax.oss.driver.core.config.DriverConfigReportingAssertions.assertDriverConfigPayload; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; @@ -30,14 +31,13 @@ import com.datastax.oss.driver.api.testinfra.session.SessionRule; import com.datastax.oss.driver.api.testinfra.session.SessionUtils; import com.datastax.oss.driver.categories.ParallelizableTests; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; +import com.datastax.oss.driver.internal.core.context.InternalDriverContext; +import com.datastax.oss.driver.internal.core.context.StartupOptionsBuilder; +import java.net.InetSocketAddress; import java.time.Duration; import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -55,7 +55,9 @@ * Simulacron only proves what the driver sends, this confirms that a real server * accepts the extra {@code STARTUP} keys and stores them, so that (a) {@code * SESSION_ID} is present on every one of the session's connections with a single shared value, and - * (b) {@code DRIVER_CONFIG} is stored for exactly one connection (the control connection). + * (b) {@code DRIVER_CONFIG} is stored for exactly one connection, which is the control connection — + * matched by address and port against the control connection's channel, so the check cannot be + * satisfied by a pooled connection. * *

Runs on both backends, asserting identical behavior — only the table that exposes the stored * options differs: ScyllaDB uses {@code system.clients.client_options}, while Apache Cassandra @@ -74,8 +76,6 @@ public class DriverConfigReportingCcmIT { private static final String DRIVER_NAME = "ScyllaDB Java Driver"; - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); - private static final CcmRule CCM_RULE = CcmRule.getInstance(); private static final SessionRule SESSION_RULE = @@ -125,48 +125,69 @@ public void should_store_session_id_on_all_connections_and_driver_config_on_cont row.getMap("client_options", String.class, String.class)); } - // (a) Every connection carries SESSION_ID, and all of them share a single value (one session). + // (a) Every row carries this session's SESSION_ID (that is what they were selected on), and + // there is more than one of them — otherwise (b) below would be vacuous. + assertThat(rows).hasSizeGreaterThanOrEqualTo(2); Set sessionIds = rows.stream().map(row -> clientOptions(row).get("SESSION_ID")).collect(Collectors.toSet()); - assertThat(sessionIds).doesNotContainNull().hasSize(1); + assertThat(sessionIds).containsExactly(sessionId(session)); - // (b) DRIVER_CONFIG is stored for exactly one connection (the control connection), and its - // value round-trips through the server intact as the stage-1 payload: valid JSON carrying - // exactly the schema version. - List driverConfigs = + // (b) DRIVER_CONFIG is stored for exactly one connection, and that connection is the control + // one — identified independently of the reported options, by the local address and port of the + // control connection's channel (which is what the server records as the client's address). + List withDriverConfig = rows.stream() - .map(row -> clientOptions(row).get("DRIVER_CONFIG")) - .filter(Objects::nonNull) + .filter(row -> clientOptions(row).get("DRIVER_CONFIG") != null) .collect(Collectors.toList()); - assertThat(driverConfigs).hasSize(1); - assertStageOnePayload(driverConfigs.get(0)); + assertThat(withDriverConfig).hasSize(1); + + Row controlRow = withDriverConfig.get(0); + InetSocketAddress controlAddress = controlConnectionAddress(session); + assertThat(controlRow.getInetAddress("address")).isEqualTo(controlAddress.getAddress()); + assertThat(controlRow.getInt("port")).isEqualTo(controlAddress.getPort()); + + // Its value round-trips through the server intact as the stage-2 payload: valid JSON carrying + // the schema version and the full configuration. + assertDriverConfigPayload(clientOptions(controlRow).get("DRIVER_CONFIG")); } /** - * Asserts that a {@code DRIVER_CONFIG} value is the stage-1 payload: well-formed JSON whose - * {@code version} is the integer {@code 1}. Guards against an incorrect schema version or a - * malformed blob slipping through a mere key-presence check. + * The local address of the control connection's channel — the source address of that TCP + * connection, and therefore what the server records in the clients table's {@code address} and + * {@code port} columns (CCM connects directly, with no proxy or address translation in between). */ - private static void assertStageOnePayload(String driverConfig) { - JsonNode root; - try { - root = OBJECT_MAPPER.readTree(driverConfig); - } catch (JsonProcessingException e) { - throw new AssertionError("DRIVER_CONFIG is not valid JSON: " + driverConfig, e); - } - assertThat(root.path("version").isInt()) - .as("version is an integer in %s", driverConfig) - .isTrue(); - assertThat(root.path("version").intValue()).isEqualTo(1); + private InetSocketAddress controlConnectionAddress(CqlSession session) { + return (InetSocketAddress) + ((InternalDriverContext) session.getContext()) + .getControlConnection() + .channel() + .localAddress(); + } + + /** + * The {@code SESSION_ID} this session reports, read from the session-wide startup options — the + * same map the driver copies into every connection's {@code STARTUP}. + */ + private String sessionId(CqlSession session) { + String sessionId = + ((InternalDriverContext) session.getContext()) + .getStartupOptions() + .get(StartupOptionsBuilder.SESSION_ID_KEY); + assertThat(sessionId).isNotNull(); + return sessionId; } /** * The rows in the clients table that belong to this driver session's connections: this driver, in - * a {@code READY} state, and carrying the reporting {@code SESSION_ID}. Transient - * protocol-version negotiation attempts (no driver identity, closed immediately) are excluded, - * and their absence here is itself the confirmation that they leave no lingering session rows. + * a {@code READY} state, and carrying this session's {@code SESSION_ID}. + * + *

Scoping on the id value matters: {@code SESSION_ID} is sent unconditionally by every driver + * session, and this class shares its CCM cluster with the other parallelizable ITs, so a + * key-presence filter would also match their connections. The {@code READY} filter is what + * excludes the transient protocol-version negotiation attempts (closed immediately). */ private List driverConnections(CqlSession session) { + String sessionId = sessionId(session); return session .execute( "SELECT address, port, connection_stage, driver_name, client_options FROM " @@ -176,7 +197,7 @@ private List driverConnections(CqlSession session) { .filter(row -> DRIVER_NAME.equals(row.getString("driver_name"))) // connection_stage casing differs across backends; compare case-insensitively. .filter(row -> "READY".equalsIgnoreCase(row.getString("connection_stage"))) - .filter(row -> clientOptions(row).containsKey("SESSION_ID")) + .filter(row -> sessionId.equals(clientOptions(row).get("SESSION_ID"))) .collect(Collectors.toList()); } diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingSimulacronIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingSimulacronIT.java index f1306fc12c9..ff20e8c54fb 100644 --- a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingSimulacronIT.java +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/DriverConfigReportingSimulacronIT.java @@ -17,9 +17,10 @@ */ package com.datastax.oss.driver.core.config; +import static com.datastax.oss.driver.core.config.DriverConfigReportingAssertions.assertDriverConfigPayload; import static com.datastax.oss.driver.internal.core.context.DefaultDriverConfigReporter.DRIVER_CONFIG_KEY; -import static com.datastax.oss.driver.internal.core.context.DefaultDriverConfigReporter.SESSION_ID_KEY; import static com.datastax.oss.driver.internal.core.context.StartupOptionsBuilder.CLIENT_ID_KEY; +import static com.datastax.oss.driver.internal.core.context.StartupOptionsBuilder.SESSION_ID_KEY; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; @@ -33,9 +34,6 @@ import com.datastax.oss.protocol.internal.request.Startup; import com.datastax.oss.simulacron.common.cluster.ClusterSpec; import com.datastax.oss.simulacron.common.cluster.QueryLog; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; import java.net.SocketAddress; import java.util.List; import java.util.Map; @@ -52,12 +50,14 @@ * CQL {@code STARTUP} frames the driver sends. * *

Simulacron records every inbound frame with its originating client connection, so we can - * verify that when {@code advanced.driver-config-reporting.enabled} is: + * verify that: * *

* *

The control connection is identified independently of the reported options: it is the only @@ -73,8 +73,6 @@ @Category(ParallelizableTests.class) public class DriverConfigReportingSimulacronIT { - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); - // A single node yields one dedicated control connection plus a pool connection (local.size // defaults to 1), i.e. at least two distinct session connections of which only the control one // registers for events. @@ -116,31 +114,14 @@ public void should_report_session_id_on_all_connections_and_driver_config_only_o assertThat(withDriverConfig).hasSize(1); assertThat(withDriverConfig.get(0).getConnection()).isEqualTo(controlConnection); - // The payload is the stage-1 report: valid JSON carrying exactly the schema version. - assertStageOnePayload(options(withDriverConfig.get(0)).get(DRIVER_CONFIG_KEY)); - } - } - - /** - * Asserts that a {@code DRIVER_CONFIG} value is the stage-1 payload: well-formed JSON whose - * {@code version} is the integer {@code 1}. Guards against an incorrect schema version or a - * malformed blob slipping through a mere key-presence check. - */ - private static void assertStageOnePayload(String driverConfig) { - JsonNode root; - try { - root = OBJECT_MAPPER.readTree(driverConfig); - } catch (JsonProcessingException e) { - throw new AssertionError("DRIVER_CONFIG is not valid JSON: " + driverConfig, e); + // The payload is the stage-2 report: valid JSON carrying the schema version and the full + // configuration (checked here via the always-present load-balancing-policy group). + assertDriverConfigPayload(options(withDriverConfig.get(0)).get(DRIVER_CONFIG_KEY)); } - assertThat(root.path("version").isInt()) - .as("version is an integer in %s", driverConfig) - .isTrue(); - assertThat(root.path("version").intValue()).isEqualTo(1); } @Test - public void should_report_nothing_when_disabled() { + public void should_still_report_session_id_when_driver_config_reporting_is_disabled() { DriverConfigLoader loader = SessionUtils.configLoaderBuilder() .withBoolean(DefaultDriverOption.DRIVER_CONFIG_REPORTING_ENABLED, false) @@ -151,13 +132,32 @@ public void should_report_nothing_when_disabled() { List startups = sessionStartups(); assertThat(distinctConnections(startups)).isGreaterThanOrEqualTo(2); - // Neither option is sent on any session connection: zero change on the wire when disabled. + // SESSION_ID does not depend on the option: it is still sent, with a single shared value... + assertThat(startups).allSatisfy(log -> assertThat(options(log)).containsKey(SESSION_ID_KEY)); + assertThat(startups.stream().map(log -> options(log).get(SESSION_ID_KEY)).distinct()) + .hasSize(1); + // ... while the configuration itself is reported nowhere. assertThat(startups) - .allSatisfy( - log -> - assertThat(options(log)) - .doesNotContainKey(SESSION_ID_KEY) - .doesNotContainKey(DRIVER_CONFIG_KEY)); + .allSatisfy(log -> assertThat(options(log)).doesNotContainKey(DRIVER_CONFIG_KEY)); + } + } + + @Test + public void should_report_driver_config_by_default() { + // No override for advanced.driver-config-reporting.enabled: exercises the shipped default. + try (CqlSession session = SessionUtils.newSession(SIMULACRON_RULE)) { + awaitControlAndPoolConnected(); + + List startups = sessionStartups(); + assertThat(distinctConnections(startups)).isGreaterThanOrEqualTo(2); + + assertThat(startups).allSatisfy(log -> assertThat(options(log)).containsKey(SESSION_ID_KEY)); + List withDriverConfig = + startups.stream() + .filter(log -> options(log).containsKey(DRIVER_CONFIG_KEY)) + .collect(Collectors.toList()); + assertThat(withDriverConfig).hasSize(1); + assertDriverConfigPayload(options(withDriverConfig.get(0)).get(DRIVER_CONFIG_KEY)); } } diff --git a/pom.xml b/pom.xml index 9a055859e6c..8499770eed7 100644 --- a/pom.xml +++ b/pom.xml @@ -93,6 +93,7 @@ 1.1.4 2.2.21 4.3.0 + 1.5.9 2.0.0-M19 3.5.5 22.0.0.2 @@ -314,6 +315,11 @@ mockito-core 5.23.0 + + com.networknt + json-schema-validator + ${json-schema-validator.version} + io.reactivex.rxjava2 rxjava diff --git a/upgrade_guide/README.md b/upgrade_guide/README.md index 214399dacc7..95a42555f53 100644 --- a/upgrade_guide/README.md +++ b/upgrade_guide/README.md @@ -19,6 +19,32 @@ under the License. ## Upgrade guide +### 4.19.2.1 + +#### The driver reports a session identifier, and its configuration, at connection time + +Two CQL `STARTUP` options are new. The server stores them in its client-connection system table +(`system.clients` on ScyllaDB, `system_views.clients` on Cassandra 4.1+), so that operators can group +a client's connections and inspect its driver settings while investigating an incident. + +* `SESSION_ID` — a driver-generated identifier, shared by all of a session's connections. It is sent + on **every** connection, unconditionally: it is an innate behavior with no configuration option to + turn it off. It is not derived from `CLIENT_ID`, which remains user-settable and unchanged. +* `DRIVER_CONFIG` — a compact JSON description of the effective configuration of the session's + default execution profile (connection/socket settings, timeouts, + retry/reconnection/speculative-execution/load-balancing policies, connection pooling, query + defaults, and TLS). Only the control connection sends it, since it describes the whole session. + It reports settings only — never credentials, statements or data — and identifies non-built-in + policies by class simple name. + +Reporting the configuration is **enabled by default**. To turn it off: + +```properties +datastax-java-driver.advanced.driver-config-reporting.enabled = false +``` + +Note that this option does not affect `SESSION_ID`. + ### 4.19.0.7 #### Cloud private-endpoint support via client routes