From 944f4be963461ba73b50dffb5a612908f6ebeb61 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 18:42:27 +0200 Subject: [PATCH 1/5] feat: client config reporting groundwork (SESSION_ID + DRIVER_CONFIG plumbing) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 1 of driver configuration reporting: the mechanism to report a client's effective configuration to ScyllaDB at connection time, so operators can inspect driver settings (via system.clients.client_options) while investigating incidents. Gated behind a new option, advanced.client-config-reporting.enabled, which ships disabled — when off there is zero change on the wire. When enabled: - SESSION_ID: a dedicated, driver-generated per-session UUID, sent on every connection (control and pool) so the server can group a session's connections. Independent of the user-settable CLIENT_ID. - DRIVER_CONFIG: a compact JSON blob, sent only on the control connection. Stage 1 emits only {"version":1} (version metadata only); the full configuration report follows in stage 2. Both entries are produced by a new DriverConfigReporter, invoked per connection from ProtocolInitHandler and evaluated fresh (so toggling the flag at runtime takes effect on new connections). Reporting is fail-safe: any failure while building the report is swallowed and never breaks connection initialization. The config option is registered consistently across DefaultDriverOption, TypedDriverOption, OptionsMap.driverDefaults() and reference.conf, and is covered by a Simulacron IT asserting the STARTUP payload. Co-Authored-By: Claude Opus 4.8 --- .../api/core/config/DefaultDriverOption.java | 17 +- .../driver/api/core/config/OptionsMap.java | 1 + .../api/core/config/TypedDriverOption.java | 5 + .../core/channel/DriverChannelOptions.java | 23 ++- .../core/channel/ProtocolInitHandler.java | 5 + .../context/DefaultDriverConfigReporter.java | 131 +++++++++++++ .../core/context/DefaultDriverContext.java | 12 ++ .../core/context/DriverConfigReporter.java | 53 ++++++ .../core/context/InternalDriverContext.java | 7 + .../core/control/ControlConnection.java | 1 + core/src/main/resources/reference.conf | 20 ++ .../core/channel/ChannelFactoryTestBase.java | 2 + .../core/channel/ProtocolInitHandlerTest.java | 68 +++++++ .../DefaultDriverConfigReporterTest.java | 141 ++++++++++++++ .../ClientConfigReportingSimulacronIT.java | 178 ++++++++++++++++++ 15 files changed, 661 insertions(+), 3 deletions(-) create mode 100644 core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java create mode 100644 core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java create mode 100644 core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java create mode 100644 integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java 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 56a6fe9c2be..9d08f6845f4 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 @@ -1173,7 +1173,22 @@ public enum DefaultDriverOption implements DriverOption { * *

Value-Type: boolean */ - ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses"); + ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses"), + /** + * Whether the driver reports its effective client 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. + * + *

Value type: boolean + */ + CLIENT_CONFIG_REPORTING_ENABLED("advanced.client-config-reporting.enabled"); private final String path; 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 60190fc1cce..86b985437cc 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,6 +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.CLIENT_CONFIG_REPORTING_ENABLED, false); } @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 e7c606cade8..e7865d449ef 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,6 +976,11 @@ public String toString() { new TypedDriverOption<>( DefaultDriverOption.CLIENT_ROUTES_SHARD_AWARENESS_ENABLED, GenericType.BOOLEAN); + /** Whether the driver reports its client configuration to ScyllaDB at connection time. */ + public static final TypedDriverOption CLIENT_CONFIG_REPORTING_ENABLED = + new TypedDriverOption<>( + DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, GenericType.BOOLEAN); + private static Iterable> introspectBuiltInValues() { try { ImmutableList.Builder> result = ImmutableList.builder(); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java index 208cf52ac22..378fd2dc0b8 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java @@ -47,15 +47,24 @@ public static Builder builder() { public final String ownerLogPrefix; + /** + * Whether the channel should report the driver configuration ({@code DRIVER_CONFIG}) in its + * {@code STARTUP} options. Set only for the control connection, so the (potentially large) config + * blob is sent once per session rather than on every pooled connection. + */ + public final boolean reportConfig; + private DriverChannelOptions( CqlIdentifier keyspace, List eventTypes, EventCallback eventCallback, - String ownerLogPrefix) { + String ownerLogPrefix, + boolean reportConfig) { this.keyspace = keyspace; this.eventTypes = eventTypes; this.eventCallback = eventCallback; this.ownerLogPrefix = ownerLogPrefix; + this.reportConfig = reportConfig; } public static class Builder { @@ -63,6 +72,7 @@ public static class Builder { private List eventTypes = Collections.emptyList(); private EventCallback eventCallback = null; private String ownerLogPrefix = null; + private boolean reportConfig = false; public Builder withKeyspace(CqlIdentifier keyspace) { this.keyspace = keyspace; @@ -82,8 +92,17 @@ public Builder withOwnerLogPrefix(String ownerLogPrefix) { return this; } + /** + * Marks this channel as the one that reports {@code DRIVER_CONFIG} (the control connection). + */ + public Builder reportConfig(boolean reportConfig) { + this.reportConfig = reportConfig; + return this; + } + public DriverChannelOptions build() { - return new DriverChannelOptions(keyspace, eventTypes, eventCallback, ownerLogPrefix); + return new DriverChannelOptions( + keyspace, eventTypes, eventCallback, ownerLogPrefix, reportConfig); } } } 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 85bffdb016e..46959cd9746 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 @@ -194,6 +194,11 @@ Message getRequest() { if (featureStore != null) { featureStore.populateStartupOptions(startupOptions); } + // Adds SESSION_ID on every connection and DRIVER_CONFIG on the control connection + // (options.reportConfig); no-op when client 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 new file mode 100644 index 00000000000..9bec4c45c7b --- /dev/null +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java @@ -0,0 +1,131 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.context; + +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.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 net.jcip.annotations.ThreadSafe; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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. + */ +@ThreadSafe +public class DefaultDriverConfigReporter implements DriverConfigReporter { + + private static final Logger LOG = LoggerFactory.getLogger(DefaultDriverConfigReporter.class); + + /** 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. + */ + static final int SCHEMA_VERSION = 1; + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + 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) { + // 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. + 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); + } + } + } catch (RuntimeException e) { + LOG.warn( + "Error while building the driver configuration report; skipping client config reporting", + e); + } + } + + private boolean isEnabled() { + return context + .getConfig() + .getDefaultProfile() + .getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false); + } + + /** + * 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. + */ + protected String buildJson() { + ObjectNode root = OBJECT_MAPPER.createObjectNode(); + root.put("version", SCHEMA_VERSION); + populateConfig(root, context.getConfig().getDefaultProfile()); + try { + return OBJECT_MAPPER.writeValueAsString(root); + } catch (JsonProcessingException e) { + // An in-memory node tree should never fail to serialize; never let it break connection setup. + LOG.warn("Failed to serialize driver configuration report; skipping DRIVER_CONFIG", e); + return null; + } + } + + /** + * 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. + */ + protected void populateConfig(ObjectNode root, DriverExecutionProfile config) { + // Stage 2: populate configuration groups from `config` and the context's policies. + } +} 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 52d9cdb264c..3d1d5b82b87 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 @@ -232,6 +232,8 @@ public class DefaultDriverContext implements InternalDriverContext { new LazyReference<>("requestThrottler", this::buildRequestThrottler, cycleDetector); private final LazyReference> startupOptionsRef = new LazyReference<>("startupOptions", this::buildStartupOptions, cycleDetector); + private final LazyReference driverConfigReporterRef = + new LazyReference<>("driverConfigReporter", this::buildDriverConfigReporter, cycleDetector); private final LazyReference nodeStateListenerRef; private final LazyReference schemaChangeListenerRef; private final LazyReference requestTrackerRef; @@ -369,6 +371,10 @@ protected Map buildStartupOptions() { .build(); } + protected DriverConfigReporter buildDriverConfigReporter() { + return new DefaultDriverConfigReporter(this); + } + protected Map buildLoadBalancingPolicies() { return Reflection.buildFromConfigProfiles( this, @@ -1226,6 +1232,12 @@ public Map getStartupOptions() { return startupOptionsRef.get(); } + @NonNull + @Override + public DriverConfigReporter getDriverConfigReporter() { + return driverConfigReporterRef.get(); + } + protected RequestLogFormatter buildRequestLogFormatter() { return new RequestLogFormatter(this); } 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 new file mode 100644 index 00000000000..d46bd3c6b71 --- /dev/null +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.context; + +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. + * + *

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

+ */ +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. + * + *

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

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. + */ + void populateStartupOptions(Map startupOptions, boolean reportDriverConfig); +} diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java index 81349b0c665..87933b7385e 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java @@ -176,6 +176,13 @@ public interface InternalDriverContext extends DriverContext { @NonNull Map getStartupOptions(); + /** + * The component that builds the driver-configuration report ({@code DRIVER_CONFIG}) sent on the + * control connection's Startup message. + */ + @NonNull + DriverConfigReporter getDriverConfigReporter(); + /** * A list of additional components to notify of session lifecycle events. * diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java b/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java index 52b32255c4b..9da3a2a8aa2 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java @@ -327,6 +327,7 @@ private void init( DriverChannelOptions.builder() .withEvents(eventTypes, ControlConnection.this) .withOwnerLogPrefix(logPrefix + "|control") + .reportConfig(true) .build(); Queue nodes = diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf index 721afa1cd76..babf2fbf2cc 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1204,6 +1204,26 @@ datastax-java-driver { } + advanced.client-config-reporting { + + # Whether the driver reports its effective client configuration to ScyllaDB 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. + # + # 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 + + } + # Whether to resolve the addresses passed to `basic.contact-points`. # # If this is true, addresses are created with `InetSocketAddress(String, int)`: the host name will 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 b25a1e9ad71..2780d5bdec9 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,6 +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) -> {}); // 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 8057ccdc48e..a7051ac466d 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 @@ -41,6 +41,7 @@ import com.datastax.oss.driver.internal.core.DefaultProtocolVersionRegistry; 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.InternalDriverContext; import com.datastax.oss.driver.internal.core.metadata.TestNodeFactory; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList; @@ -102,6 +103,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. + when(internalDriverContext.getDriverConfigReporter()) + .thenReturn((startupOptions, reportDriverConfig) -> {}); channel .pipeline() @@ -153,6 +157,70 @@ public void should_initialize() { assertThat(connectFuture).isSuccess(); } + // Mirrors the real reporter: SESSION_ID on every connection, DRIVER_CONFIG only when asked. + private void stubConfigReporter() { + when(internalDriverContext.getDriverConfigReporter()) + .thenReturn( + (startupOptions, reportDriverConfig) -> { + startupOptions.put(DefaultDriverConfigReporter.SESSION_ID_KEY, "test-session-id"); + if (reportDriverConfig) { + startupOptions.put( + DefaultDriverConfigReporter.DRIVER_CONFIG_KEY, "{\"version\":1}"); + } + }); + } + + @Test + public void should_report_session_id_and_driver_config_on_control_connection() { + stubConfigReporter(); + channel + .pipeline() + .addLast( + ChannelFactory.INIT_HANDLER_NAME, + new ProtocolInitHandler( + internalDriverContext, + DefaultProtocolVersion.V4, + null, + END_POINT, + DriverChannelOptions.builder().reportConfig(true).build(), + 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).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(); + channel + .pipeline() + .addLast( + ChannelFactory.INIT_HANDLER_NAME, + new ProtocolInitHandler( + internalDriverContext, + DefaultProtocolVersion.V4, + null, + END_POINT, + // DriverChannelOptions.DEFAULT has reportConfig = false (a pool connection). + 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).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(startup.options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + @Test public void should_query_supported_options() { channel 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 new file mode 100644 index 00000000000..3624a94d41f --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.context; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +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.DriverExecutionProfile; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import org.junit.Before; +import org.junit.Test; + +public class DefaultDriverConfigReporterTest { + + private InternalDriverContext context; + private DriverExecutionProfile profile; + private DefaultDriverConfigReporter reporter; + + @Before + public void setup() { + context = 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); + } + + private void enableReporting(boolean enabled) { + when(profile.getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false)) + .thenReturn(enabled); + } + + @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); + } + + @Test + public void should_add_session_id_and_driver_config_on_control_connection() { + enableReporting(true); + 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 + "}"); + } + + @Test + public void should_add_session_id_only_on_pool_connection() { + enableReporting(true); + Map options = new HashMap<>(); + reporter.populateStartupOptions(options, /* reportDriverConfig= */ false); + assertThat(options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + + @Test + public void should_use_a_stable_session_id_across_connections() { + 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)); + } + + @Test + public void should_use_a_distinct_session_id_per_reporter() { + 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)); + } + + /** 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.CLIENT_CONFIG_REPORTING_ENABLED, false)) + .thenThrow(new IllegalStateException("config blew up")); + 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); + } + + /** + * 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. + */ + @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); + } +} diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java new file mode 100644 index 00000000000..eb3d30fab71 --- /dev/null +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java @@ -0,0 +1,178 @@ +/* + * 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 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 org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.testinfra.session.SessionUtils; +import com.datastax.oss.driver.api.testinfra.simulacron.SimulacronRule; +import com.datastax.oss.driver.categories.ParallelizableTests; +import com.datastax.oss.protocol.internal.request.Register; +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 java.net.SocketAddress; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * End-to-end check of client-configuration reporting against a mock server, asserting on the actual + * 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.client-config-reporting.enabled} is: + * + *

+ * + *

The control connection is identified independently of the reported options: it is the only + * connection that issues a {@code REGISTER} frame (to subscribe to cluster events). + * + *

Assertions are scoped to the session's real connections, identified by the {@code CLIENT_ID} + * startup option that the driver always sends (independently of config reporting). This excludes + * the short-lived connections opened by protocol-version negotiation: when the protocol version is + * not pinned (the default), the driver first tries higher versions (DSE_V2, DSE_V1, V5) that + * ScyllaDB rejects before the handshake completes, and Simulacron records each such rejected + * attempt as a bare {@code STARTUP} ({@code CQL_VERSION} only) that carries no driver identity. + */ +@Category(ParallelizableTests.class) +public class ClientConfigReportingSimulacronIT { + + // 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. + @ClassRule + public static final SimulacronRule SIMULACRON_RULE = + new SimulacronRule(ClusterSpec.builder().withNodes(1)); + + @Before + public void clearLogs() { + SIMULACRON_RULE.cluster().clearLogs(); + } + + @Test + public void should_report_session_id_on_all_connections_and_driver_config_only_on_control() { + DriverConfigLoader loader = + SessionUtils.configLoaderBuilder() + .withBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, true) + .build(); + try (CqlSession session = SessionUtils.newSession(SIMULACRON_RULE, loader)) { + awaitControlAndPoolConnected(); + + SocketAddress controlConnection = controlConnection().orElseThrow(AssertionError::new); + List startups = sessionStartups(); + + // Sanity: we are actually observing more than one connection (control + at least one pool), + // otherwise the "only on the control connection" assertion below would be vacuous. + assertThat(distinctConnections(startups)).isGreaterThanOrEqualTo(2); + + // SESSION_ID is present on every session connection and has the same value everywhere. + 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); + + // DRIVER_CONFIG is present on exactly one connection, and that connection is the control one. + List withDriverConfig = + startups.stream() + .filter(log -> options(log).containsKey(DRIVER_CONFIG_KEY)) + .collect(Collectors.toList()); + assertThat(withDriverConfig).hasSize(1); + assertThat(withDriverConfig.get(0).getConnection()).isEqualTo(controlConnection); + } + } + + @Test + public void should_report_nothing_when_disabled() { + DriverConfigLoader loader = + SessionUtils.configLoaderBuilder() + .withBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false) + .build(); + try (CqlSession session = SessionUtils.newSession(SIMULACRON_RULE, loader)) { + awaitControlAndPoolConnected(); + + List startups = sessionStartups(); + assertThat(distinctConnections(startups)).isGreaterThanOrEqualTo(2); + + // Neither option is sent on any session connection: zero change on the wire when disabled. + assertThat(startups) + .allSatisfy( + log -> + assertThat(options(log)) + .doesNotContainKey(SESSION_ID_KEY) + .doesNotContainKey(DRIVER_CONFIG_KEY)); + } + } + + /** + * The {@code STARTUP} frames of the session's real connections (control + pool), identified by + * the always-present {@code CLIENT_ID} option; excludes protocol-version negotiation attempts. + */ + private List sessionStartups() { + return SIMULACRON_RULE.cluster().getLogs().getQueryLogs().stream() + .filter(log -> log.getFrame().message instanceof Startup) + .filter(log -> options(log).containsKey(CLIENT_ID_KEY)) + .collect(Collectors.toList()); + } + + /** The {@code STARTUP} options carried by a recorded frame. */ + private Map options(QueryLog log) { + return ((Startup) log.getFrame().message).options; + } + + /** The connection that issued a {@code REGISTER} frame, i.e. the control connection. */ + private Optional controlConnection() { + return SIMULACRON_RULE.cluster().getLogs().getQueryLogs().stream() + .filter(log -> log.getFrame().message instanceof Register) + .map(QueryLog::getConnection) + .findFirst(); + } + + private long distinctConnections(List logs) { + return logs.stream().map(QueryLog::getConnection).distinct().count(); + } + + /** + * Waits until both the control connection (identified by its {@code REGISTER}) and at least one + * pool connection have sent their {@code STARTUP}, since pool connections may finish initializing + * slightly after the session builder returns. + */ + private void awaitControlAndPoolConnected() { + await() + .pollInterval(100, TimeUnit.MILLISECONDS) + .atMost(30, TimeUnit.SECONDS) + .until( + () -> controlConnection().isPresent() && distinctConnections(sessionStartups()) >= 2); + } +} From 9aa1a5642210164e1382faf19cdc16bfc6b22e4e Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Wed, 22 Jul 2026 17:05:08 +0200 Subject: [PATCH 2/5] feat: populate the full DRIVER_CONFIG report (stage 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fills in DefaultDriverConfigReporter with all configuration groups of the cross-driver schema, replacing the stage-1 {"version":1} placeholder: - connection, socket, control_pane_settings: mapped from the default profile (timeouts in ms; keys with no Java equivalent — read/write timeout, server-side internal-query timeout — reported as null). - reconnection_policy / retry_policy / speculative_execution_policy: a discriminated object chosen by the configured policy class, with a "custom" fallback carrying the class name for user policies. - load_balancing_policy: normalized flags for DefaultLoadBalancingPolicy (token-aware, dc-failover, explicit/inferred local dc, ...), "custom" otherwise. - connection_pool, query_defaults, tls: pool sizing / shard-aware port, per-request defaults (page size, consistency, idempotence, client timestamps), and TLS booleans only (never credentials or hosts). Values are read from the default execution profile and the context's policy accessors on each control-connection init, so the report reflects the current (possibly reloaded) configuration. Building remains fail-safe. Adds unit coverage for the default report shape and each policy discrimination branch, driven by OptionsMap.driverDefaults(). Co-Authored-By: Claude Opus 4.8 --- .../context/DefaultDriverConfigReporter.java | 255 +++++++++++- .../DefaultDriverConfigReporterTest.java | 364 ++++++++++++++++-- 2 files changed, 575 insertions(+), 44 deletions(-) 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 9bec4c45c7b..e6d85a6ef37 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 @@ -19,9 +19,24 @@ 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.config.DriverOption; +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.uuid.Uuids; +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.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.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.NullNode; import com.fasterxml.jackson.databind.node.ObjectNode; import java.util.Map; import java.util.UUID; @@ -121,11 +136,243 @@ 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; keys the Java driver + * has no equivalent for are reported as {@code null} rather than dropped. */ protected void populateConfig(ObjectNode root, DriverExecutionProfile config) { - // Stage 2: populate configuration groups from `config` and the context's policies. + root.set("connection", connection(config)); + root.set("socket", socket(config)); + root.set("control_pane_settings", controlPlaneSettings(config)); + root.set("reconnection_policy", reconnectionPolicy(config)); + root.set("retry_policy", retryPolicy()); + root.set("speculative_execution_policy", speculativeExecutionPolicy(config)); + root.set("load_balancing_policy", loadBalancingPolicy(config)); + // The Java driver has no session-level DC/rack preference API separate from the LB policy. + root.putNull("node_location_preference"); + root.set("connection_pool", connectionPool(config)); + root.set("query_defaults", queryDefaults(config)); + root.set("tls", tls(config)); + } + + private JsonNode connection(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put( + "connect_timeout_ms", + config.getDuration(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT).toMillis()); + // The Java driver has no socket-level read/write timeouts. + n.putNull("read_timeout_ms"); + n.putNull("write_timeout_ms"); + putMillisOrNull(n, "cql_heartbeat_interval_ms", config, DefaultDriverOption.HEARTBEAT_INTERVAL); + putMillisOrNull(n, "cql_heartbeat_timeout_ms", config, DefaultDriverOption.HEARTBEAT_TIMEOUT); + return n; + } + + private JsonNode socket(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("tcp_no_delay", config.getBoolean(DefaultDriverOption.SOCKET_TCP_NODELAY, true)); + // keep_alive / reuse_address are not set by default; report the effective driver value (the OS + // default, which the driver leaves untouched, is approximated as false when unset). + n.put("keep_alive", config.getBoolean(DefaultDriverOption.SOCKET_KEEP_ALIVE, false)); + n.put("reuse_address", config.getBoolean(DefaultDriverOption.SOCKET_REUSE_ADDRESS, false)); + putIntOrNull(n, "linger_interval_s", config, DefaultDriverOption.SOCKET_LINGER_INTERVAL); + putIntOrNull(n, "receive_buffer_size", config, DefaultDriverOption.SOCKET_RECEIVE_BUFFER_SIZE); + putIntOrNull(n, "send_buffer_size", config, DefaultDriverOption.SOCKET_SEND_BUFFER_SIZE); + return n; + } + + private JsonNode controlPlaneSettings(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put( + "system_query_client_side_timeout_ms", + config.getDuration(DefaultDriverOption.CONTROL_CONNECTION_TIMEOUT).toMillis()); + // The Java metadata/control timeout is client-side; the driver does not add USING TIMEOUT, so + // the server-side field is not applicable. + n.putNull("system_query_server_side_timeout_ms"); + n.put( + "schema_agreement_timeout_ms", + config.getDuration(DefaultDriverOption.CONTROL_CONNECTION_AGREEMENT_TIMEOUT).toMillis()); + return n; + } + + private JsonNode reconnectionPolicy(DriverExecutionProfile config) { + ReconnectionPolicy policy = context.getReconnectionPolicy(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy instanceof ExponentialReconnectionPolicy) { + n.put("type", "exponential"); + n.put( + "base_delay_ms", + config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); + n.put( + "max_delay_ms", + config.getDuration(DefaultDriverOption.RECONNECTION_MAX_DELAY).toMillis()); + // Java's built-in reconnection policies are unbounded. + n.putNull("max_attempts"); + } else if (policy instanceof ConstantReconnectionPolicy) { + n.put("type", "constant"); + n.put( + "fixed_delay_ms", + config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); + n.putNull("max_attempts"); + } else { + customPolicy(n, policy); + } + return n; + } + + private JsonNode retryPolicy() { + RetryPolicy policy = context.getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy instanceof DefaultRetryPolicy) { + n.put("type", "standard-error-aware"); + n.putNull("backoff"); + } else if (policy instanceof ConsistencyDowngradingRetryPolicy) { + n.put("type", "downgrading-consistency"); + n.putNull("backoff"); + } else { + customPolicy(n, policy); + } + return n; + } + + private JsonNode speculativeExecutionPolicy(DriverExecutionProfile config) { + SpeculativeExecutionPolicy policy = + context.getSpeculativeExecutionPolicy(DriverExecutionProfile.DEFAULT_NAME); + if (policy instanceof NoSpeculativeExecutionPolicy) { + return NullNode.getInstance(); + } + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy instanceof ConstantSpeculativeExecutionPolicy) { + 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 JsonNode loadBalancingPolicy(DriverExecutionProfile config) { + LoadBalancingPolicy policy = + context.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy instanceof DefaultLoadBalancingPolicy) { + n.put("type", "default"); + // The Java default policy is always token-aware and has no replica-shuffle option. + n.put("token_aware", true); + n.put("shuffle", false); + // DC is inferred from the first contacted node when not configured explicitly; rack has no + // preference when unset. + n.set( + "local_dc", locality(config, DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, true)); + n.set("local_rack", locality(config, DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK, false)); + n.put( + "dc_failover", + config.getInt(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_MAX_NODES_PER_REMOTE_DC, 0) + > 0); + n.put("latency_awareness", false); + } else { + customPolicy(n, policy); + } + return n; + } + + private JsonNode connectionPool(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("type", "host"); + n.put( + "desired_connections_count", config.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE)); + putIntOrNull( + n, "max_requests_per_connection", config, DefaultDriverOption.CONNECTION_MAX_REQUESTS); + n.put( + "shard_aware_port_enabled", + config.getBoolean(DefaultDriverOption.CONNECTION_ADVANCED_SHARD_AWARENESS_ENABLED, true)); + return n; + } + + private JsonNode queryDefaults(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + int pageSize = config.getInt(DefaultDriverOption.REQUEST_PAGE_SIZE); + if (pageSize <= 0) { + n.put("page_size", "unbounded"); + } else { + n.put("page_size", pageSize); + } + n.put("consistency", config.getString(DefaultDriverOption.REQUEST_CONSISTENCY)); + putStringOrNull( + n, "serial_consistency", config, 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)); + putMillisOrNull(n, "request_timeout_ms", config, DefaultDriverOption.REQUEST_TIMEOUT); + return n; + } + + private JsonNode tls(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + boolean enabled = context.getSslEngineFactory().isPresent(); + n.put("enabled", enabled); + n.put( + "hostname_verification", + enabled && config.getBoolean(DefaultDriverOption.SSL_HOSTNAME_VALIDATION, true)); + return n; + } + + /** + * A datacenter/rack locality: {@code {source, value}} when configured explicitly, {@code + * {source:"inferred", value:null}} when absent but the driver still applies a preference (DC), or + * {@code null} when there is no preference at all (rack). + */ + private JsonNode locality( + DriverExecutionProfile config, DriverOption option, boolean inferredWhenAbsent) { + if (config.isDefined(option)) { + ObjectNode loc = OBJECT_MAPPER.createObjectNode(); + loc.put("source", "explicit"); + loc.put("value", config.getString(option)); + return loc; + } + if (inferredWhenAbsent) { + ObjectNode loc = OBJECT_MAPPER.createObjectNode(); + loc.put("source", "inferred"); + loc.putNull("value"); + return loc; + } + return NullNode.getInstance(); + } + + private void customPolicy(ObjectNode node, Object policy) { + node.put("type", "custom"); + node.put("name", policy.getClass().getSimpleName()); + } + + private void putMillisOrNull( + ObjectNode node, String key, DriverExecutionProfile config, DriverOption option) { + if (config.isDefined(option)) { + node.put(key, config.getDuration(option).toMillis()); + } else { + node.putNull(key); + } + } + + private void putIntOrNull( + ObjectNode node, String key, DriverExecutionProfile config, DriverOption option) { + if (config.isDefined(option)) { + node.put(key, config.getInt(option)); + } else { + node.putNull(key); + } + } + + private void putStringOrNull( + ObjectNode node, String key, DriverExecutionProfile config, DriverOption option) { + if (config.isDefined(option)) { + node.put(key, config.getString(option)); + } else { + node.putNull(key); + } } } 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 3624a94d41f..fe4e0fb56db 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 @@ -23,35 +23,61 @@ 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.fasterxml.jackson.databind.node.ObjectNode; +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.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.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.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 java.time.Duration; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.UUID; +import java.util.function.Consumer; import org.junit.Before; import org.junit.Test; public class DefaultDriverConfigReporterTest { - private InternalDriverContext context; - private DriverExecutionProfile profile; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + // ---- 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.CLIENT_CONFIG_REPORTING_ENABLED, false)) + when(mockProfile.getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false)) .thenReturn(enabled); } + // ==================== Gating ==================== + @Test public void should_add_nothing_when_disabled() { enableReporting(false); @@ -61,26 +87,14 @@ public void should_add_nothing_when_disabled() { assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); } - @Test - public void should_add_session_id_and_driver_config_on_control_connection() { - enableReporting(true); - 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 + "}"); - } - @Test public void should_add_session_id_only_on_pool_connection() { enableReporting(true); Map options = new HashMap<>(); reporter.populateStartupOptions(options, /* reportDriverConfig= */ false); - assertThat(options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + String sessionId = options.get(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(sessionId).isNotNull(); + assertThat(UUID.fromString(sessionId)).isNotNull(); // valid UUID assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); } @@ -89,7 +103,7 @@ public void should_use_a_stable_session_id_across_connections() { enableReporting(true); Map control = new HashMap<>(); Map pool = new HashMap<>(); - reporter.populateStartupOptions(control, true); + reporter.populateStartupOptions(control, false); reporter.populateStartupOptions(pool, false); assertThat(pool.get(DefaultDriverConfigReporter.SESSION_ID_KEY)) .isEqualTo(control.get(DefaultDriverConfigReporter.SESSION_ID_KEY)); @@ -100,17 +114,17 @@ public void should_use_a_distinct_session_id_per_reporter() { 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); + new DefaultDriverConfigReporter(mockContext).populateStartupOptions(second, false); assertThat(second.get(DefaultDriverConfigReporter.SESSION_ID_KEY)) .isNotEqualTo(first.get(DefaultDriverConfigReporter.SESSION_ID_KEY)); } - /** Reporting must never break the connection: a failed config read is swallowed entirely. */ + // ==================== Fail-safe ==================== + @Test public void should_not_throw_when_reading_the_flag_fails() { - when(profile.getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false)) + when(mockProfile.getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false)) .thenThrow(new IllegalStateException("config blew up")); Map options = new HashMap<>(); reporter.populateStartupOptions(options, true); // must not throw @@ -118,24 +132,294 @@ public void should_not_throw_when_reading_the_flag_fails() { assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); } - /** - * 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. - */ @Test - public void should_keep_session_id_but_skip_driver_config_when_building_config_groups_fails() { + public void should_keep_session_id_but_skip_driver_config_when_building_fails() { enableReporting(true); - DefaultDriverConfigReporter throwingReporter = - new DefaultDriverConfigReporter(context) { + DefaultDriverConfigReporter throwing = + new DefaultDriverConfigReporter(mockContext) { @Override - protected void populateConfig(ObjectNode root, DriverExecutionProfile config) { - throw new IllegalStateException("policy introspection blew up"); + protected String buildJson() { + throw new IllegalStateException("introspection blew up"); } }; Map options = new HashMap<>(); - throwingReporter.populateStartupOptions(options, true); // must not throw + throwing.populateStartupOptions(options, true); // must not throw assertThat(options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); } + + // ==================== Report content ==================== + + @Test + public void should_report_default_configuration() throws Exception { + JsonNode report = report(defaultsReporter(map -> {})); + + assertThat(report.get("version").asInt()).isEqualTo(DefaultDriverConfigReporter.SCHEMA_VERSION); + + // All groups present. + for (String group : + new String[] { + "connection", + "socket", + "control_pane_settings", + "reconnection_policy", + "retry_policy", + "speculative_execution_policy", + "load_balancing_policy", + "node_location_preference", + "connection_pool", + "query_defaults", + "tls" + }) { + assertThat(report.has(group)).as("group %s present", group).isTrue(); + } + + JsonNode connection = report.get("connection"); + assertThat(connection.get("connect_timeout_ms").asLong()).isPositive(); + assertThat(connection.get("read_timeout_ms").isNull()).isTrue(); + assertThat(connection.get("write_timeout_ms").isNull()).isTrue(); + + JsonNode socket = report.get("socket"); + assertThat(socket.get("tcp_no_delay").asBoolean()).isTrue(); + assertThat(socket.get("keep_alive").asBoolean()).isFalse(); + assertThat(socket.get("linger_interval_s").isNull()).isTrue(); + assertThat(socket.get("receive_buffer_size").isNull()).isTrue(); + + // Java-specific reconciliation: no server-side (USING TIMEOUT) internal-query timeout. + assertThat( + report.get("control_pane_settings").get("system_query_server_side_timeout_ms").isNull()) + .isTrue(); + + JsonNode reconnection = report.get("reconnection_policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("exponential"); + assertThat(reconnection.get("base_delay_ms").asLong()).isPositive(); + assertThat(reconnection.get("max_attempts").isNull()).isTrue(); + + assertThat(report.get("retry_policy").get("type").asText()).isEqualTo("standard-error-aware"); + + // Default speculative execution policy is "none" => null. + assertThat(report.get("speculative_execution_policy").isNull()).isTrue(); + + JsonNode lb = report.get("load_balancing_policy"); + assertThat(lb.get("type").asText()).isEqualTo("default"); + assertThat(lb.get("token_aware").asBoolean()).isTrue(); + assertThat(lb.get("shuffle").asBoolean()).isFalse(); + assertThat(lb.get("latency_awareness").asBoolean()).isFalse(); + assertThat(lb.get("dc_failover").asBoolean()).isFalse(); + // local-datacenter not configured in the defaults => inferred, value not yet known. + assertThat(lb.get("local_dc").get("source").asText()).isEqualTo("inferred"); + assertThat(lb.get("local_dc").get("value").isNull()).isTrue(); + assertThat(lb.get("local_rack").isNull()).isTrue(); + + assertThat(report.get("node_location_preference").isNull()).isTrue(); + + 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_port_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_timeout_ms").asLong()).isPositive(); + + JsonNode tls = report.get("tls"); + assertThat(tls.get("enabled").asBoolean()).isFalse(); + assertThat(tls.get("hostname_verification").asBoolean()).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("fixed_delay_ms").asLong()).isPositive(); + assertThat(reconnection.get("max_attempts").isNull()).isTrue(); + } + + @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_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_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_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 localDc = report(r).get("load_balancing_policy").get("local_dc"); + assertThat(localDc.get("source").asText()).isEqualTo("explicit"); + assertThat(localDc.get("value").asText()).isEqualTo("dc1"); + } + + @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 { + 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(mock(SslEngineFactory.class))); + JsonNode tls = report(r).get("tls"); + assertThat(tls.get("enabled").asBoolean()).isTrue(); + assertThat(tls.get("hostname_verification").asBoolean()).isTrue(); + } + + @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_size").asInt()).isEqualTo(65535); + assertThat(socket.get("linger_interval_s").asInt()).isEqualTo(5); + } + + @Test + public void should_report_unbounded_page_size() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.REQUEST_PAGE_SIZE, 0)); + assertThat(report(r).get("query_defaults").get("page_size").asText()).isEqualTo("unbounded"); + } + + // ==================== helpers ==================== + + private JsonNode report(DefaultDriverConfigReporter reporter) throws Exception { + return MAPPER.readTree(reporter.buildJson()); + } + + /** 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) { + 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); + return new DefaultDriverConfigReporter(ctx); + } } From 7c6eb1aa45a23541a3652185137bc9f160bb8a83 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 17:33:37 +0200 Subject: [PATCH 3/5] test: add live-ScyllaDB IT for client config reporting (stage 2) ClientConfigReportingCcmIT is the real-server counterpart of the Simulacron test: it starts ScyllaDB via CCM, connects with client config reporting enabled, and reads back system.clients.client_options to assert what the server actually stored: - SESSION_ID is present on every one of the session's connections, with a single shared value; - DRIVER_CONFIG is stored for exactly one connection (the control connection), identified independently as the connection carrying it. Verified against ScyllaDB 2026.1.9: the server accepts the extra STARTUP keys and stores the full DRIVER_CONFIG report intact (no length truncation), and transient protocol-version negotiation attempts leave no rows in system.clients. Scylla-only via @BackendRequirement; runs in the parallelizable CCM suite. Co-Authored-By: Claude Opus 4.8 --- .../config/ClientConfigReportingCcmIT.java | 140 ++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java new file mode 100644 index 00000000000..7280390d9cc --- /dev/null +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java @@ -0,0 +1,140 @@ +/* + * 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 static org.awaitility.Awaitility.await; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.cql.Row; +import com.datastax.oss.driver.api.testinfra.ccm.CcmRule; +import com.datastax.oss.driver.api.testinfra.requirement.BackendRequirement; +import com.datastax.oss.driver.api.testinfra.requirement.BackendType; +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 java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.rules.RuleChain; +import org.junit.rules.TestRule; + +/** + * Verifies client-configuration reporting end-to-end against a live ScyllaDB (via CCM), by reading + * back what the server actually stored in {@code system.clients.client_options}. + * + *

This is the real-server counterpart of {@code ClientConfigReportingSimulacronIT}: it confirms + * (a) that ScyllaDB accepts the extra {@code STARTUP} keys and stores them, (b) that {@code + * SESSION_ID} is present on every one of the session's connections with a single shared value, and + * (c) that {@code DRIVER_CONFIG} is stored for exactly one connection (the control connection). + * + *

ScyllaDB-only: {@code system.clients.client_options} is a Scylla feature. + */ +@Category(ParallelizableTests.class) +@BackendRequirement( + type = BackendType.SCYLLA, + description = "system.clients.client_options is a ScyllaDB feature") +public class ClientConfigReportingCcmIT { + + private static final String DRIVER_NAME = "ScyllaDB Java Driver"; + + private static final CcmRule CCM_RULE = CcmRule.getInstance(); + + private static final SessionRule SESSION_RULE = + SessionRule.builder(CCM_RULE) + .withConfigLoader( + SessionUtils.configLoaderBuilder() + .withBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, true) + .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(30)) + .build()) + .build(); + + @ClassRule + public static final TestRule CHAIN = RuleChain.outerRule(CCM_RULE).around(SESSION_RULE); + + @Test + public void should_store_session_id_on_all_connections_and_driver_config_on_control() { + CqlSession session = SESSION_RULE.session(); + + // The session opens a control connection plus at least one pooled connection; wait until the + // server reflects them (system.clients is updated asynchronously as connections are set up). + await() + .atMost(60, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .until(() -> driverConnections(session).size() >= 2); + + List rows = driverConnections(session); + + // Evidence for the record: dump exactly what the server stored per connection. + System.out.println("system.clients rows for this driver session:"); + for (Row row : rows) { + System.out.printf( + " %s:%s stage=%s client_options=%s%n", + row.getObject("address"), + row.getObject("port"), + row.getString("connection_stage"), + row.getMap("client_options", String.class, String.class)); + } + + // (b) Every connection carries SESSION_ID, and all of them share a single value (one session). + Set sessionIds = + rows.stream().map(row -> clientOptions(row).get("SESSION_ID")).collect(Collectors.toSet()); + assertThat(sessionIds).doesNotContainNull().hasSize(1); + + // (c) DRIVER_CONFIG is stored for exactly one connection (the control connection), and it is + // the + // compact JSON blob the driver reports. + List driverConfigs = + rows.stream() + .map(row -> clientOptions(row).get("DRIVER_CONFIG")) + .filter(java.util.Objects::nonNull) + .collect(Collectors.toList()); + assertThat(driverConfigs).hasSize(1); + assertThat(driverConfigs.get(0)).contains("\"version\""); + } + + /** + * The rows in {@code system.clients} 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. + */ + private List driverConnections(CqlSession session) { + return session + .execute( + "SELECT address, port, connection_stage, driver_name, client_options FROM system.clients") + .all() + .stream() + .filter(row -> DRIVER_NAME.equals(row.getString("driver_name"))) + .filter(row -> "READY".equals(row.getString("connection_stage"))) + .filter(row -> clientOptions(row).containsKey("SESSION_ID")) + .collect(Collectors.toList()); + } + + private Map clientOptions(Row row) { + Map options = row.getMap("client_options", String.class, String.class); + return options == null ? java.util.Collections.emptyMap() : options; + } +} From 607af6f1126ee12883aca4368a1f5096d8b95a4a Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 21:31:35 +0200 Subject: [PATCH 4/5] test: gate client-config-reporting CCM IT to ScyllaDB 2026.1+ ClientConfigReportingCcmIT reads system.clients.client_options, a column that only exists on newer ScyllaDB releases. On the LTS-PRIOR lane (Scylla 2025.1.14) the column is absent, so the SELECT fails with InvalidQueryException: "Unrecognized name client_options", failing the "Scylla ITs (LTS-PRIOR, 11, parallelizable)" check (the test carries @Category(ParallelizableTests.class), so only that lane runs it). Add a minInclusive="2026.1.0" bound to the existing @BackendRequirement so the test is skipped on versions without the column and still runs on 2026.1+ (LTS-LATEST 2026.1.9 and LATEST already pass). Co-Authored-By: Claude Opus 4.8 --- .../oss/driver/core/config/ClientConfigReportingCcmIT.java | 1 + 1 file changed, 1 insertion(+) diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java index 7280390d9cc..a4546b2efa6 100644 --- a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java @@ -55,6 +55,7 @@ @Category(ParallelizableTests.class) @BackendRequirement( type = BackendType.SCYLLA, + minInclusive = "2026.1.0", description = "system.clients.client_options is a ScyllaDB feature") public class ClientConfigReportingCcmIT { From ffe06098429ed4c04d0cd56e2ffd1f4afdb2cd9c Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Fri, 24 Jul 2026 18:40:17 +0200 Subject: [PATCH 5/5] fix: correct policy misclassification and locality resolution in DRIVER_CONFIG report Code review of the stage-2 (full report) work surfaced several correctness bugs: - node-location-preference ignored SessionBuilder.withLocalDatacenter(), only checking the config option, and silently dropped an explicitly-configured rack when no DC was set (now reported as rack-auto instead of being lost). - Policy discrimination used instanceof against non-final built-in classes, so DcInferringLoadBalancingPolicy (a real, documented policy) was misreported as "default", and any user subclass of a built-in reconnection/retry/speculative-execution/load-balancing policy was misreported as the unmodified built-in instead of "custom". Switched all four to exact-class checks; DcInferringLoadBalancingPolicy now gets its own "dc-inferring" type. - customPolicy() reported an empty name for anonymous policy classes; falls back to the binary class name now. - CLIENT_CONFIG_REPORTING_ENABLED's javadoc/reference.conf comment still described the stage-1 {"version":1}-only placeholder. - Corrected a stale rationale comment in ClientConfigReportingSimulacronIT. Added test coverage for each fix, and documented (rather than worked around) remaining known limitations: single-execution-profile scoping and LazyReference init-ordering assumptions. Co-Authored-By: Claude Sonnet 5 --- .../api/core/config/DefaultDriverOption.java | 8 +- .../context/DefaultDriverConfigReporter.java | 339 ++++++++++-------- core/src/main/resources/reference.conf | 9 +- .../DefaultDriverConfigReporterTest.java | 321 +++++++++++++---- .../ClientConfigReportingSimulacronIT.java | 5 +- 5 files changed, 466 insertions(+), 216 deletions(-) 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 9d08f6845f4..24bf038a7fd 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 @@ -1181,10 +1181,10 @@ public enum DefaultDriverOption implements DriverOption { * 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. + * key on the control connection only, describing 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). + * When {@code false}, neither entry is sent and there is no change on the wire. * *

Value type: boolean */ 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 e6d85a6ef37..9e66601123d 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 @@ -19,7 +19,6 @@ 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.config.DriverOption; 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; @@ -27,6 +26,7 @@ import com.datastax.oss.driver.api.core.uuid.Uuids; 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.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; @@ -34,9 +34,7 @@ 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.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.node.NullNode; import com.fasterxml.jackson.databind.node.ObjectNode; import java.util.Map; import java.util.UUID; @@ -50,6 +48,24 @@ * *

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. + * + *

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, but {@link #buildJson()} and + * {@link #populateConfig(ObjectNode, DriverExecutionProfile)} are {@code protected} purely to let + * tests override them (see {@code DefaultDriverConfigReporterTest}); a production subclass that + * overrides either to add caching or other mutable state is responsible for its own synchronization + * — {@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 { @@ -119,8 +135,12 @@ private boolean isEnabled() { /** * 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. */ protected String buildJson() { ObjectNode root = OBJECT_MAPPER.createObjectNode(); @@ -137,116 +157,142 @@ protected String buildJson() { /** * 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; keys the Java driver - * has no equivalent for are reported as {@code null} rather than dropped. + * 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}. */ protected void populateConfig(ObjectNode root, DriverExecutionProfile config) { root.set("connection", connection(config)); root.set("socket", socket(config)); - root.set("control_pane_settings", controlPlaneSettings(config)); - root.set("reconnection_policy", reconnectionPolicy(config)); - root.set("retry_policy", retryPolicy()); - root.set("speculative_execution_policy", speculativeExecutionPolicy(config)); - root.set("load_balancing_policy", loadBalancingPolicy(config)); - // The Java driver has no session-level DC/rack preference API separate from the LB policy. - root.putNull("node_location_preference"); - root.set("connection_pool", connectionPool(config)); - root.set("query_defaults", queryDefaults(config)); + root.set("control-plane", controlPlane(config)); + 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(config)); } - private JsonNode connection(DriverExecutionProfile config) { + private ObjectNode connection(DriverExecutionProfile config) { ObjectNode n = OBJECT_MAPPER.createObjectNode(); - n.put( - "connect_timeout_ms", + ObjectNode connect = OBJECT_MAPPER.createObjectNode(); + connect.put( + "timeout-ms", config.getDuration(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT).toMillis()); - // The Java driver has no socket-level read/write timeouts. - n.putNull("read_timeout_ms"); - n.putNull("write_timeout_ms"); - putMillisOrNull(n, "cql_heartbeat_interval_ms", config, DefaultDriverOption.HEARTBEAT_INTERVAL); - putMillisOrNull(n, "cql_heartbeat_timeout_ms", config, DefaultDriverOption.HEARTBEAT_TIMEOUT); + 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 JsonNode socket(DriverExecutionProfile config) { + private ObjectNode socket(DriverExecutionProfile config) { ObjectNode n = OBJECT_MAPPER.createObjectNode(); - n.put("tcp_no_delay", config.getBoolean(DefaultDriverOption.SOCKET_TCP_NODELAY, true)); - // keep_alive / reuse_address are not set by default; report the effective driver value (the OS + n.put("tcp-no-delay", config.getBoolean(DefaultDriverOption.SOCKET_TCP_NODELAY, true)); + // keep-alive / reuse-address are not set by default; report the effective driver value (the OS // default, which the driver leaves untouched, is approximated as false when unset). - n.put("keep_alive", config.getBoolean(DefaultDriverOption.SOCKET_KEEP_ALIVE, false)); - n.put("reuse_address", config.getBoolean(DefaultDriverOption.SOCKET_REUSE_ADDRESS, false)); - putIntOrNull(n, "linger_interval_s", config, DefaultDriverOption.SOCKET_LINGER_INTERVAL); - putIntOrNull(n, "receive_buffer_size", config, DefaultDriverOption.SOCKET_RECEIVE_BUFFER_SIZE); - putIntOrNull(n, "send_buffer_size", config, DefaultDriverOption.SOCKET_SEND_BUFFER_SIZE); + 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 JsonNode controlPlaneSettings(DriverExecutionProfile config) { + private ObjectNode controlPlane(DriverExecutionProfile config) { ObjectNode n = OBJECT_MAPPER.createObjectNode(); - n.put( - "system_query_client_side_timeout_ms", + + ObjectNode timeout = OBJECT_MAPPER.createObjectNode(); + timeout.put( + "client-side-ms", config.getDuration(DefaultDriverOption.CONTROL_CONNECTION_TIMEOUT).toMillis()); // The Java metadata/control timeout is client-side; the driver does not add USING TIMEOUT, so - // the server-side field is not applicable. - n.putNull("system_query_server_side_timeout_ms"); - n.put( - "schema_agreement_timeout_ms", + // the server-side field is omitted (not applicable). + 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 JsonNode reconnectionPolicy(DriverExecutionProfile config) { + private ObjectNode reconnectionPolicy(DriverExecutionProfile config) { ReconnectionPolicy policy = context.getReconnectionPolicy(); ObjectNode n = OBJECT_MAPPER.createObjectNode(); - if (policy instanceof ExponentialReconnectionPolicy) { + // 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_delay_ms", - config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); - n.put( - "max_delay_ms", - config.getDuration(DefaultDriverOption.RECONNECTION_MAX_DELAY).toMillis()); - // Java's built-in reconnection policies are unbounded. - n.putNull("max_attempts"); - } else if (policy instanceof ConstantReconnectionPolicy) { + 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( - "fixed_delay_ms", - config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); - n.putNull("max_attempts"); + n.put("delay-ms", config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); } else { customPolicy(n, policy); } return n; } - private JsonNode retryPolicy() { + private ObjectNode retryPolicy() { RetryPolicy policy = context.getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME); ObjectNode n = OBJECT_MAPPER.createObjectNode(); - if (policy instanceof DefaultRetryPolicy) { + // 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"); - n.putNull("backoff"); - } else if (policy instanceof ConsistencyDowngradingRetryPolicy) { + // No configurable backoff: omitted. + } else if (policy.getClass() == ConsistencyDowngradingRetryPolicy.class) { n.put("type", "downgrading-consistency"); - n.putNull("backoff"); + // No configurable backoff: omitted. } else { customPolicy(n, policy); } return n; } - private JsonNode speculativeExecutionPolicy(DriverExecutionProfile config) { + /** + * 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); - if (policy instanceof NoSpeculativeExecutionPolicy) { - return NullNode.getInstance(); + // Exact-class checks, not instanceof: neither built-in is final. + if (policy.getClass() == NoSpeculativeExecutionPolicy.class) { + return null; } ObjectNode n = OBJECT_MAPPER.createObjectNode(); - if (policy instanceof ConstantSpeculativeExecutionPolicy) { + if (policy.getClass() == ConstantSpeculativeExecutionPolicy.class) { n.put("type", "constant"); - n.put("max_executions", config.getInt(DefaultDriverOption.SPECULATIVE_EXECUTION_MAX)); + n.put("max-executions", config.getInt(DefaultDriverOption.SPECULATIVE_EXECUTION_MAX)); n.put( - "delay_ms", + "delay-ms", config.getDuration(DefaultDriverOption.SPECULATIVE_EXECUTION_DELAY).toMillis()); } else { customPolicy(n, policy); @@ -254,125 +300,132 @@ private JsonNode speculativeExecutionPolicy(DriverExecutionProfile config) { return n; } - private JsonNode loadBalancingPolicy(DriverExecutionProfile config) { + private ObjectNode loadBalancingPolicy(DriverExecutionProfile config) { LoadBalancingPolicy policy = context.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME); ObjectNode n = OBJECT_MAPPER.createObjectNode(); - if (policy instanceof DefaultLoadBalancingPolicy) { - n.put("type", "default"); + // DcInferringLoadBalancingPolicy extends DefaultLoadBalancingPolicy, overriding only how the + // local DC is discovered; every flag below is unaffected, so the two share a branch and differ + // only in the reported "type". Exact-class checks (not instanceof) so an actual user subclass + // of either still falls through to "custom" below. + if (policy.getClass() == DefaultLoadBalancingPolicy.class + || policy.getClass() == DcInferringLoadBalancingPolicy.class) { + n.put( + "type", + policy.getClass() == DcInferringLoadBalancingPolicy.class ? "dc-inferring" : "default"); // The Java default policy is always token-aware and has no replica-shuffle option. - n.put("token_aware", true); + n.put("token-aware", true); n.put("shuffle", false); - // DC is inferred from the first contacted node when not configured explicitly; rack has no - // preference when unset. - n.set( - "local_dc", locality(config, DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, true)); - n.set("local_rack", locality(config, DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK, false)); n.put( - "dc_failover", + "dc-failover", config.getInt(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_MAX_NODES_PER_REMOTE_DC, 0) > 0); - n.put("latency_awareness", false); + n.put("latency-awareness", false); } else { customPolicy(n, policy); } return n; } - private JsonNode connectionPool(DriverExecutionProfile config) { + /** + * 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. + */ + private ObjectNode nodeLocationPreference(DriverExecutionProfile config) { + String localDc = context.getLocalDatacenter(DriverExecutionProfile.DEFAULT_NAME); + boolean dcExplicit = localDc != null; + if (!dcExplicit && config.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) { + localDc = config.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER); + dcExplicit = true; + } + boolean rackExplicit = config.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK); + String localRack = + rackExplicit ? config.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK) : null; + + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (dcExplicit && rackExplicit) { + n.put("type", "rack"); + n.put("local-dc", localDc); + n.put("local-rack", localRack); + } else if (dcExplicit) { + n.put("type", "dc"); + n.put("local-dc", localDc); + } else if (rackExplicit) { + // 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)); - putIntOrNull( - n, "max_requests_per_connection", config, DefaultDriverOption.CONNECTION_MAX_REQUESTS); - n.put( - "shard_aware_port_enabled", + "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 JsonNode queryDefaults(DriverExecutionProfile config) { + private ObjectNode queryDefaults(DriverExecutionProfile config) { ObjectNode n = OBJECT_MAPPER.createObjectNode(); int pageSize = config.getInt(DefaultDriverOption.REQUEST_PAGE_SIZE); - if (pageSize <= 0) { - n.put("page_size", "unbounded"); - } else { - n.put("page_size", pageSize); + 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)); - putStringOrNull( - n, "serial_consistency", config, DefaultDriverOption.REQUEST_SERIAL_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", + "client-timestamps", !(context.getTimestampGenerator() instanceof ServerSideTimestampGenerator)); - putMillisOrNull(n, "request_timeout_ms", config, DefaultDriverOption.REQUEST_TIMEOUT); + ObjectNode request = OBJECT_MAPPER.createObjectNode(); + request.put("timeout-ms", config.getDuration(DefaultDriverOption.REQUEST_TIMEOUT).toMillis()); + n.set("request", request); return n; } - private JsonNode tls(DriverExecutionProfile config) { + private ObjectNode tls(DriverExecutionProfile config) { ObjectNode n = OBJECT_MAPPER.createObjectNode(); boolean enabled = context.getSslEngineFactory().isPresent(); n.put("enabled", enabled); n.put( - "hostname_verification", + "hostname-verification", enabled && config.getBoolean(DefaultDriverOption.SSL_HOSTNAME_VALIDATION, true)); return n; } - /** - * A datacenter/rack locality: {@code {source, value}} when configured explicitly, {@code - * {source:"inferred", value:null}} when absent but the driver still applies a preference (DC), or - * {@code null} when there is no preference at all (rack). - */ - private JsonNode locality( - DriverExecutionProfile config, DriverOption option, boolean inferredWhenAbsent) { - if (config.isDefined(option)) { - ObjectNode loc = OBJECT_MAPPER.createObjectNode(); - loc.put("source", "explicit"); - loc.put("value", config.getString(option)); - return loc; - } - if (inferredWhenAbsent) { - ObjectNode loc = OBJECT_MAPPER.createObjectNode(); - loc.put("source", "inferred"); - loc.putNull("value"); - return loc; - } - return NullNode.getInstance(); - } - private void customPolicy(ObjectNode node, Object policy) { node.put("type", "custom"); - node.put("name", policy.getClass().getSimpleName()); - } - - private void putMillisOrNull( - ObjectNode node, String key, DriverExecutionProfile config, DriverOption option) { - if (config.isDefined(option)) { - node.put(key, config.getDuration(option).toMillis()); - } else { - node.putNull(key); - } - } - - private void putIntOrNull( - ObjectNode node, String key, DriverExecutionProfile config, DriverOption option) { - if (config.isDefined(option)) { - node.put(key, config.getInt(option)); - } else { - node.putNull(key); - } - } - - private void putStringOrNull( - ObjectNode node, String key, DriverExecutionProfile config, DriverOption option) { - if (config.isDefined(option)) { - node.put(key, config.getString(option)); - } else { - node.putNull(key); - } + // 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/resources/reference.conf b/core/src/main/resources/reference.conf index babf2fbf2cc..a93fe42772c 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1211,10 +1211,11 @@ datastax-java-driver { # 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. + # and a compact JSON payload under the DRIVER_CONFIG key on the control connection only, + # describing 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). When false, neither entry is sent and there is + # no change on the wire. # # Required: no # Modifiable at runtime: yes, the new value will be used for connections initialized after the change. 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 fe4e0fb56db..79a85a19f1a 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 @@ -28,6 +28,7 @@ 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; @@ -35,6 +36,7 @@ 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.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; @@ -156,77 +158,90 @@ public void should_report_default_configuration() throws Exception { assertThat(report.get("version").asInt()).isEqualTo(DefaultDriverConfigReporter.SCHEMA_VERSION); - // All groups present. + // Groups always present for the default profile. for (String group : new String[] { "connection", "socket", - "control_pane_settings", - "reconnection_policy", - "retry_policy", - "speculative_execution_policy", - "load_balancing_policy", - "node_location_preference", - "connection_pool", - "query_defaults", + "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_timeout_ms").asLong()).isPositive(); - assertThat(connection.get("read_timeout_ms").isNull()).isTrue(); - assertThat(connection.get("write_timeout_ms").isNull()).isTrue(); + 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.get("linger_interval_s").isNull()).isTrue(); - assertThat(socket.get("receive_buffer_size").isNull()).isTrue(); + 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(); // Java-specific reconciliation: no server-side (USING TIMEOUT) internal-query timeout. - assertThat( - report.get("control_pane_settings").get("system_query_server_side_timeout_ms").isNull()) - .isTrue(); + 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"); + JsonNode reconnection = report.get("reconnection-policy"); assertThat(reconnection.get("type").asText()).isEqualTo("exponential"); - assertThat(reconnection.get("base_delay_ms").asLong()).isPositive(); - assertThat(reconnection.get("max_attempts").isNull()).isTrue(); + 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(); - assertThat(report.get("retry_policy").get("type").asText()).isEqualTo("standard-error-aware"); + JsonNode retry = report.get("retry-policy"); + assertThat(retry.get("type").asText()).isEqualTo("standard-error-aware"); + assertThat(retry.has("backoff")).isFalse(); - // Default speculative execution policy is "none" => null. - assertThat(report.get("speculative_execution_policy").isNull()).isTrue(); - - JsonNode lb = report.get("load_balancing_policy"); + JsonNode lb = report.get("load-balancing-policy"); assertThat(lb.get("type").asText()).isEqualTo("default"); - assertThat(lb.get("token_aware").asBoolean()).isTrue(); + assertThat(lb.get("token-aware").asBoolean()).isTrue(); assertThat(lb.get("shuffle").asBoolean()).isFalse(); - assertThat(lb.get("latency_awareness").asBoolean()).isFalse(); - assertThat(lb.get("dc_failover").asBoolean()).isFalse(); - // local-datacenter not configured in the defaults => inferred, value not yet known. - assertThat(lb.get("local_dc").get("source").asText()).isEqualTo("inferred"); - assertThat(lb.get("local_dc").get("value").isNull()).isTrue(); - assertThat(lb.get("local_rack").isNull()).isTrue(); - - assertThat(report.get("node_location_preference").isNull()).isTrue(); - - JsonNode pool = report.get("connection_pool"); + assertThat(lb.get("latency-awareness").asBoolean()).isFalse(); + 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_port_enabled").asBoolean()).isTrue(); + assertThat(pool.get("desired-connections-count").asInt()).isPositive(); + assertThat(pool.get("shard-aware").get("enabled").asBoolean()).isTrue(); - JsonNode query = report.get("query_defaults"); + 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_timeout_ms").asLong()).isPositive(); + 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(); + assertThat(tls.get("hostname-verification").asBoolean()).isFalse(); } @Test @@ -240,10 +255,10 @@ public void should_report_constant_reconnection_policy() throws Exception { mock(DefaultLoadBalancingPolicy.class), mock(TimestampGenerator.class), Optional.empty()); - JsonNode reconnection = report(r).get("reconnection_policy"); + JsonNode reconnection = report(r).get("reconnection-policy"); assertThat(reconnection.get("type").asText()).isEqualTo("constant"); - assertThat(reconnection.get("fixed_delay_ms").asLong()).isPositive(); - assertThat(reconnection.get("max_attempts").isNull()).isTrue(); + assertThat(reconnection.get("delay-ms").asLong()).isPositive(); + assertThat(reconnection.has("max-attempts")).isFalse(); } @Test @@ -257,8 +272,31 @@ public void should_report_custom_reconnection_policy() throws Exception { mock(DefaultLoadBalancingPolicy.class), mock(TimestampGenerator.class), Optional.empty()); - JsonNode reconnection = report(r).get("reconnection_policy"); + 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(); } @@ -273,10 +311,29 @@ public void should_report_downgrading_consistency_retry_policy() throws Exceptio mock(DefaultLoadBalancingPolicy.class), mock(TimestampGenerator.class), Optional.empty()); - assertThat(report(r).get("retry_policy").get("type").asText()) + 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 = @@ -292,10 +349,48 @@ public void should_report_constant_speculative_execution_policy() throws Excepti mock(DefaultLoadBalancingPolicy.class), mock(TimestampGenerator.class), Optional.empty()); - JsonNode spec = report(r).get("speculative_execution_policy"); + 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); + 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(); } @Test @@ -309,7 +404,7 @@ public void should_report_custom_load_balancing_policy() throws Exception { mock(LoadBalancingPolicy.class), // not the default policy mock(TimestampGenerator.class), Optional.empty()); - JsonNode lb = report(r).get("load_balancing_policy"); + JsonNode lb = report(r).get("load-balancing-policy"); assertThat(lb.get("type").asText()).isEqualTo("custom"); assertThat(lb.get("name").asText()).isNotEmpty(); } @@ -318,9 +413,73 @@ public void should_report_custom_load_balancing_policy() throws Exception { public void should_report_explicit_local_dc() throws Exception { DefaultDriverConfigReporter r = defaultsReporter(map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc1")); - JsonNode localDc = report(r).get("load_balancing_policy").get("local_dc"); - assertThat(localDc.get("source").asText()).isEqualTo("explicit"); - assertThat(localDc.get("value").asText()).isEqualTo("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 @@ -335,7 +494,7 @@ public void should_report_server_side_timestamps_as_disabled_client_timestamps() mock(DefaultLoadBalancingPolicy.class), mock(ServerSideTimestampGenerator.class), Optional.empty()); - assertThat(report(r).get("query_defaults").get("client_timestamps").asBoolean()).isFalse(); + assertThat(report(r).get("query-defaults").get("client-timestamps").asBoolean()).isFalse(); } @Test @@ -351,7 +510,7 @@ public void should_report_tls_enabled_with_hostname_verification() throws Except Optional.of(mock(SslEngineFactory.class))); JsonNode tls = report(r).get("tls"); assertThat(tls.get("enabled").asBoolean()).isTrue(); - assertThat(tls.get("hostname_verification").asBoolean()).isTrue(); + assertThat(tls.get("hostname-verification").asBoolean()).isTrue(); } @Test @@ -364,16 +523,24 @@ public void should_report_socket_overrides() throws Exception { 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_size").asInt()).isEqualTo(65535); - assertThat(socket.get("linger_interval_s").asInt()).isEqualTo(5); + 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_report_unbounded_page_size() throws Exception { + public void should_omit_page_when_unbounded() throws Exception { DefaultDriverConfigReporter r = defaultsReporter(map -> map.put(TypedDriverOption.REQUEST_PAGE_SIZE, 0)); - assertThat(report(r).get("query_defaults").get("page_size").asText()).isEqualTo("unbounded"); + // 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); } // ==================== helpers ==================== @@ -409,6 +576,20 @@ private DefaultDriverConfigReporter reporterWith( 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); @@ -420,6 +601,20 @@ private DefaultDriverConfigReporter reporterWith( 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/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java index eb3d30fab71..1003686313a 100644 --- a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java @@ -64,8 +64,9 @@ * startup option that the driver always sends (independently of config reporting). This excludes * the short-lived connections opened by protocol-version negotiation: when the protocol version is * not pinned (the default), the driver first tries higher versions (DSE_V2, DSE_V1, V5) that - * ScyllaDB rejects before the handshake completes, and Simulacron records each such rejected - * attempt as a bare {@code STARTUP} ({@code CQL_VERSION} only) that carries no driver identity. + * ScyllaDB rejects; those attempts are torn down during negotiation before the driver ever reaches + * the point of sending a {@code STARTUP} that carries its full identifying options (including + * {@code CLIENT_ID}), so filtering on {@code CLIENT_ID}'s presence naturally excludes them. */ @Category(ParallelizableTests.class) public class ClientConfigReportingSimulacronIT {