diff --git a/driver-core/pom.xml b/driver-core/pom.xml index 66a9ba3afb2..8af93c038de 100644 --- a/driver-core/pom.xml +++ b/driver-core/pom.xml @@ -164,6 +164,7 @@ + com.fasterxml.jackson.core jackson-core @@ -172,10 +173,6 @@ com.fasterxml.jackson.core jackson-databind - - com.fasterxml.jackson.dataformat - jackson-dataformat-yaml - diff --git a/driver-core/src/main/java/com/datastax/driver/core/Cluster.java b/driver-core/src/main/java/com/datastax/driver/core/Cluster.java index 0cc335c9d70..817fce39f31 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/Cluster.java +++ b/driver-core/src/main/java/com/datastax/driver/core/Cluster.java @@ -1443,6 +1443,26 @@ public Builder withApplicationInfo(ApplicationInfo applicationInfo) { return this; } + /** + * Enables or disables driver configuration reporting, i.e. whether the control connection sends + * a {@code DRIVER_CONFIG} JSON blob describing the effective driver configuration in its + * startup options. Enabled by default. + * + *

The server stores it in {@code system.clients.client_options} — a per-node table, so only + * the node holding the control connection stores {@code DRIVER_CONFIG}; consumers must query + * and aggregate across all nodes. + * + *

This does not govern the {@code SESSION_ID} startup option, which every connection always + * sends (so the server can group every connection opened from this {@link Cluster}, across all + * of its {@link Session}s), regardless of this setting. + * + * @param enabled whether driver configuration reporting is enabled. + */ + public Builder withDriverConfigReporting(boolean enabled) { + configurationBuilder.withDriverConfigReporting(enabled); + return this; + } + /** * The configuration that will be used for the new cluster. * @@ -1598,6 +1618,7 @@ private Manager( .withNettyOptions(configuration.getNettyOptions()) .withCodecRegistry(configuration.getCodecRegistry()) .withApplicationInfo(configuration.getApplicationInfo()) + .withDriverConfigReporting(configuration.isDriverConfigReportingEnabled()) .build(); } else { this.configuration = configuration; diff --git a/driver-core/src/main/java/com/datastax/driver/core/Configuration.java b/driver-core/src/main/java/com/datastax/driver/core/Configuration.java index e26c0292505..354d76f2b31 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/Configuration.java +++ b/driver-core/src/main/java/com/datastax/driver/core/Configuration.java @@ -60,6 +60,7 @@ public static Builder builder() { private final CodecRegistry codecRegistry; private final String defaultKeyspace; private final ApplicationInfo applicationInfo; + private final boolean driverConfigReportingEnabled; private Configuration( Policies policies, @@ -72,7 +73,8 @@ private Configuration( NettyOptions nettyOptions, CodecRegistry codecRegistry, String defaultKeyspace, - ApplicationInfo applicationInfo) { + ApplicationInfo applicationInfo, + boolean driverConfigReportingEnabled) { this.policies = policies; this.protocolOptions = protocolOptions; this.poolingOptions = poolingOptions; @@ -84,6 +86,7 @@ private Configuration( this.codecRegistry = codecRegistry; this.defaultKeyspace = defaultKeyspace; this.applicationInfo = applicationInfo; + this.driverConfigReportingEnabled = driverConfigReportingEnabled; } /** @@ -103,7 +106,8 @@ protected Configuration(Configuration toCopy) { toCopy.getNettyOptions(), toCopy.getCodecRegistry(), toCopy.getDefaultKeyspace(), - toCopy.getApplicationInfo()); + toCopy.getApplicationInfo(), + toCopy.isDriverConfigReportingEnabled()); } void register(Cluster.Manager manager) { @@ -222,6 +226,20 @@ public ApplicationInfo getApplicationInfo() { return applicationInfo; } + /** + * Whether driver configuration reporting is enabled, i.e. whether the control connection sends a + * {@code DRIVER_CONFIG} JSON blob describing the effective driver configuration in its startup + * options. Enabled by default. + * + *

This does not govern the {@code SESSION_ID} startup option, which every connection always + * sends regardless of this setting. + * + * @return {@code true} if driver configuration reporting is enabled. + */ + public boolean isDriverConfigReportingEnabled() { + return driverConfigReportingEnabled; + } + /** * Returns the {@link CodecRegistry} instance for this configuration. * @@ -247,6 +265,7 @@ public static class Builder { private ThreadingOptions threadingOptions; private NettyOptions nettyOptions; private ApplicationInfo applicationInfo; + private boolean driverConfigReportingEnabled = true; private CodecRegistry codecRegistry; private String defaultKeyspace; @@ -261,6 +280,19 @@ public Builder withApplicationInfo(ApplicationInfo applicationInfo) { return this; } + /** + * Enables or disables driver configuration reporting (the {@code DRIVER_CONFIG} startup option + * sent by the control connection). Enabled by default; see {@link + * Configuration#isDriverConfigReportingEnabled()}. + * + * @param driverConfigReportingEnabled whether driver configuration reporting is enabled. + * @return this builder. + */ + public Builder withDriverConfigReporting(boolean driverConfigReportingEnabled) { + this.driverConfigReportingEnabled = driverConfigReportingEnabled; + return this; + } + /** * Sets the policies for this cluster. * @@ -392,7 +424,8 @@ public Configuration build() { nettyOptions != null ? nettyOptions : NettyOptions.DEFAULT_INSTANCE, codecRegistry != null ? codecRegistry : CodecRegistry.DEFAULT_INSTANCE, defaultKeyspace, - applicationInfo); + applicationInfo, + driverConfigReportingEnabled); } } } diff --git a/driver-core/src/main/java/com/datastax/driver/core/Connection.java b/driver-core/src/main/java/com/datastax/driver/core/Connection.java index 798b04912c7..c8f021bf299 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/Connection.java +++ b/driver-core/src/main/java/com/datastax/driver/core/Connection.java @@ -87,6 +87,7 @@ import java.util.List; import java.util.Map; import java.util.Queue; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; @@ -118,6 +119,12 @@ class Connection { private static final long ADV_SHARD_AWARENESS_BLOCK_ON_ERROR = 5 * 60 * 1000; + /** + * STARTUP option key under which the Cluster-scoped connection-grouping identifier is sent, on + * every connection (see {@link Factory#sessionId}). + */ + static final String SESSION_ID_KEY = "SESSION_ID"; + enum State { OPEN, TRASHED, @@ -159,6 +166,11 @@ enum State { private final ApplicationInfo applicationInfo; private ProtocolFeatureStore protocolFeatureStore; + // The DRIVER_CONFIG blob this connection reports in its STARTUP options, or null to report none. + // Set only for the control connection, so the (potentially large) config blob is sent once per + // Cluster rather than on every connection. SESSION_ID is still sent on every connection. + private final String driverConfig; + /** * Create a new connection to a Cassandra node and associate it with the given pool. * @@ -169,6 +181,11 @@ enum State { * connection can also be associated to an owner later with {@link #setOwner(Owner)}. */ protected Connection(String name, EndPoint endPoint, Factory factory, Owner owner) { + this(name, endPoint, factory, owner, null); + } + + private Connection( + String name, EndPoint endPoint, Factory factory, Owner owner, String driverConfig) { this.endPoint = endPoint; this.factory = factory; this.dispatcher = new Dispatcher(); @@ -178,6 +195,7 @@ protected Connection(String name, EndPoint endPoint, Factory factory, Owner owne this.defaultKeyspaceAttempt = new SetKeyspaceAttempt(null, thisFuture); this.targetKeyspace = new AtomicReference(defaultKeyspaceAttempt); this.applicationInfo = factory.configuration.getApplicationInfo(); + this.driverConfig = driverConfig; } /** Create a new connection to a Cassandra node. */ @@ -512,6 +530,16 @@ public ListenableFuture apply(Void input) throws Exception { applicationInfo.addOption(extraOptions); } + // Sent on every connection, unconditionally (like DRIVER_NAME / DRIVER_VERSION), so the + // server can group all the connections opened from this Cluster. + extraOptions.put(SESSION_ID_KEY, factory.sessionId.toString()); + + // Built once when the Cluster initialized; non-null only on the control connection, and + // only when driver config reporting is enabled. + if (driverConfig != null) { + extraOptions.put(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY, driverConfig); + } + if (protocolFeatureStore != null) { protocolFeatureStore.populateStartupOptions(protocolVersion, extraOptions); } @@ -1283,11 +1311,28 @@ static class Factory { volatile ProtocolVersion protocolVersion; private final NettyOptions nettyOptions; + // Dedicated, driver-generated identifier sent as SESSION_ID on every connection, so the server + // can group them. Not derived from the (user-settable) CLIENT_ID, so that it is guaranteed + // unique as the grouping key requires. There is one Factory per Cluster, so this is stable and + // shared by every Session obtained from that Cluster (the control connection has no Session + // affiliation, so Cluster-wide is the finest granularity available here). + final UUID sessionId = UUID.randomUUID(); + + // The DRIVER_CONFIG blob sent on the control connection, or null when driver config reporting + // is disabled (or the report could not be built). Built once here, as the Cluster initializes, + // and reused for every control connection this factory opens: it is never rebuilt while the + // session is in flight. + final String driverConfig; + Factory(Cluster.Manager manager, Configuration configuration) { this.defaultHandler = manager; this.manager = manager; this.reaper = manager.reaper; this.configuration = configuration; + this.driverConfig = + configuration.isDriverConfigReportingEnabled() + ? new DefaultDriverConfigReporter(configuration).buildReport() + : null; this.authProvider = configuration.getProtocolOptions().getAuthProvider(); this.protocolVersion = configuration.getProtocolOptions().initialProtocolVersion; this.nettyOptions = configuration.getNettyOptions(); @@ -1319,12 +1364,25 @@ int getPort() { Connection open(Host host) throws ConnectionException, InterruptedException, UnsupportedProtocolVersionException, ClusterNameMismatchException { + return open(host, false); + } + + /** + * Same as {@link #open(Host)}, but when {@code reportConfig} is true, hands the connection the + * {@code DRIVER_CONFIG} blob to report, marking it as the control connection (a no-op when + * driver config reporting is disabled, since there is then no blob to report). + */ + Connection open(Host host, boolean reportConfig) + throws ConnectionException, InterruptedException, UnsupportedProtocolVersionException, + ClusterNameMismatchException { EndPoint endPoint = host.getEndPoint(); if (isShutdown) throw new ConnectionException(endPoint, "Connection factory is shut down"); host.convictionPolicy.signalConnectionsOpening(1); - Connection connection = new Connection(buildConnectionName(host), endPoint, this); + Connection connection = + new Connection( + buildConnectionName(host), endPoint, this, null, reportConfig ? driverConfig : null); // This method opens the connection synchronously, so wait until it's initialized try { connection.initAsync().get(); diff --git a/driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java b/driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java index d72b82e2052..da178813482 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java +++ b/driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java @@ -322,7 +322,9 @@ private static Map logError( private Connection tryConnect(Host host, boolean isInitialConnection) throws ConnectionException, ExecutionException, InterruptedException, UnsupportedProtocolVersionException, ClusterNameMismatchException { - Connection connection = cluster.connectionFactory.open(host); + // Mark the control connection so it reports the full DRIVER_CONFIG blob (other connections send + // only SESSION_ID). No-op unless driver config reporting is enabled. + Connection connection = cluster.connectionFactory.open(host, true); String productType = connection.optionsQuery().get(); // If no protocol version was specified, set the default as soon as a connection succeeds (it's // needed to parse UDTs in refreshSchema) diff --git a/driver-core/src/main/java/com/datastax/driver/core/DefaultDriverConfigReporter.java b/driver-core/src/main/java/com/datastax/driver/core/DefaultDriverConfigReporter.java new file mode 100644 index 00000000000..30c01848608 --- /dev/null +++ b/driver-core/src/main/java/com/datastax/driver/core/DefaultDriverConfigReporter.java @@ -0,0 +1,98 @@ +/* + * Copyright ScyllaDB, Inc. + * + * Licensed 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.driver.core; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Default {@link DriverConfigReporter}: serializes the driver configuration to the cross-driver + * {@code DRIVER_CONFIG} JSON shape, which {@link Connection.Factory} then sends in the control + * connection's {@code STARTUP} options. + * + *

The report is built once, when the {@link Cluster} initializes, and the resulting string is + * reused for the lifetime of that {@code Cluster} — it is never rebuilt while the session is in + * flight, so a control-connection reconnect costs nothing and always reports the same + * configuration. + */ +public class DefaultDriverConfigReporter implements DriverConfigReporter { + + private static final Logger LOGGER = LoggerFactory.getLogger(DefaultDriverConfigReporter.class); + + /** STARTUP option key under which the config JSON is sent. */ + public static final String DRIVER_CONFIG_KEY = "DRIVER_CONFIG"; + + /** + * 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 Configuration configuration; + + public DefaultDriverConfigReporter(Configuration configuration) { + this.configuration = configuration; + } + + @Override + public String buildReport() { + // Configuration reporting is a best-effort diagnostic aid, so any failure here (a bad config + // read, a misbehaving policy while introspecting, a serialization error) must be swallowed + // rather than allowed to propagate: it is built on the Cluster-initialization path, which must + // not fail because of a diagnostic. + try { + return buildJson(); + } catch (RuntimeException e) { + LOGGER.warn( + "Error while building the driver configuration report; skipping driver config reporting", + e); + return null; + } + } + + /** + * 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)} in a later stage. + */ + protected String buildJson() { + ObjectNode root = OBJECT_MAPPER.createObjectNode(); + root.put("version", SCHEMA_VERSION); + populateConfig(root); + 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. + LOGGER.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. from {@link #configuration}. + */ + protected void populateConfig(ObjectNode root) { + // Stage 2: populate configuration groups from `configuration`. + } +} diff --git a/driver-core/src/main/java/com/datastax/driver/core/DriverConfigReporter.java b/driver-core/src/main/java/com/datastax/driver/core/DriverConfigReporter.java new file mode 100644 index 00000000000..7ada1c34a2b --- /dev/null +++ b/driver-core/src/main/java/com/datastax/driver/core/DriverConfigReporter.java @@ -0,0 +1,48 @@ +/* + * Copyright ScyllaDB, Inc. + * + * Licensed 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.driver.core; + +/** + * Builds the {@code DRIVER_CONFIG} payload that the control connection sends in its CQL {@code + * STARTUP} options, so ScyllaDB can store it in {@code system.clients.client_options} and operators + * can inspect a client's effective driver settings while investigating incidents. + * + *

Only the control connection carries the blob; pooled connections are correlated back to it via + * the {@code SESSION_ID} startup option, which {@link Connection} sends on every connection + * independently of this reporter and of {@link Cluster.Builder#withDriverConfigReporting(boolean)}. + * + *

{@code system.clients.client_options} is per node: only the node holding the control + * connection stores {@code DRIVER_CONFIG}; other nodes only see {@code SESSION_ID}-bearing pooled- + * connection rows. Consumers must query and aggregate across all nodes to see the full picture. + */ +public interface DriverConfigReporter { + + /** + * Builds the configuration report, or returns {@code null} if it could not be built (in which + * case no {@code DRIVER_CONFIG} option is sent). + * + *

Called once per {@link Cluster}, while it initializes; the returned string is then reused + * for every control connection that cluster opens. + * + *

Implementations must not throw: a failure to build the report must be swallowed (and + * logged) rather than propagated, so that a diagnostic aid can never break cluster + * initialization. + * + * @return the report to send under the {@code DRIVER_CONFIG} startup option, or {@code null} to + * send nothing. + */ + String buildReport(); +} diff --git a/driver-core/src/test/java/com/datastax/driver/core/DefaultDriverConfigReporterTest.java b/driver-core/src/test/java/com/datastax/driver/core/DefaultDriverConfigReporterTest.java new file mode 100644 index 00000000000..cabc6a591a5 --- /dev/null +++ b/driver-core/src/test/java/com/datastax/driver/core/DefaultDriverConfigReporterTest.java @@ -0,0 +1,60 @@ +/* + * Copyright ScyllaDB, Inc. + * + * Licensed 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.driver.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.testng.annotations.Test; + +public class DefaultDriverConfigReporterTest { + + private static Configuration config() { + return Configuration.builder().build(); + } + + @Test(groups = "unit") + public void should_enable_driver_config_reporting_by_default() { + assertThat(config().isDriverConfigReportingEnabled()).isTrue(); + assertThat( + Cluster.builder() + .addContactPoint("127.0.0.1") + .getConfiguration() + .isDriverConfigReportingEnabled()) + .isTrue(); + } + + @Test(groups = "unit") + public void should_report_schema_version() { + // Stage 1 emits only the schema version. + assertThat(new DefaultDriverConfigReporter(config()).buildReport()) + .isEqualTo("{\"version\":1}"); + } + + @Test(groups = "unit") + public void should_be_fail_safe_when_report_build_throws() { + DefaultDriverConfigReporter reporter = + new DefaultDriverConfigReporter(config()) { + @Override + protected String buildJson() { + throw new RuntimeException("boom"); + } + }; + + // Must not propagate the failure (it runs on the cluster-initialization path); no report means + // no DRIVER_CONFIG option is sent, and nothing else about the connection is affected. + assertThat(reporter.buildReport()).isNull(); + } +} diff --git a/driver-core/src/test/java/com/datastax/driver/core/DriverConfigReportingCcmTest.java b/driver-core/src/test/java/com/datastax/driver/core/DriverConfigReportingCcmTest.java new file mode 100644 index 00000000000..a51ed8568d3 --- /dev/null +++ b/driver-core/src/test/java/com/datastax/driver/core/DriverConfigReportingCcmTest.java @@ -0,0 +1,233 @@ +/* + * Copyright ScyllaDB, Inc. + * + * Licensed 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.driver.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.datastax.driver.core.utils.ScyllaVersion; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.testng.annotations.Test; + +/** + * Verifies driver-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}: + * + *

+ * + *

The cluster under test uses the default configuration — no {@code withDriverConfigReporting} + * call — so these also assert that reporting is enabled by default. + * + *

Stage 1 emits only the schema version, so {@code DRIVER_CONFIG} is asserted to be {@code + * {"version":1}}. ScyllaDB-only: {@code system.clients.client_options} is a Scylla feature (added + * in ScyllaDB 2026.1). + */ +@ScyllaVersion( + minOSS = "2026.1", + minEnterprise = "2026.1", + description = "system.clients.client_options requires ScyllaDB 2026.1+") +public class DriverConfigReportingCcmTest extends CCMTestsSupport { + + private static final String DRIVER_NAME = "ScyllaDB Java Driver"; + + @Test(groups = "short") + public void should_store_session_id_on_all_connections_and_driver_config_on_control() { + String sessionId = sessionId(cluster()); + + // The session opens a control connection plus at least one pooled connection; system.clients is + // updated asynchronously as connections are set up, so poll until the server reflects them. + List rows = awaitClusterConnections(sessionId, 2); + // Guard against the "only on control connection" assertion below being vacuously true because + // the pool connection never showed up (e.g. it was slow, or SESSION_ID reporting regressed on + // it) rather than because reporting is actually correct. + assertThat(rows.size()) + .as("connections carrying this cluster's SESSION_ID") + .isGreaterThanOrEqualTo(2); + + List driverConfigs = new ArrayList(); + for (Row row : rows) { + String driverConfig = clientOptions(row).get("DRIVER_CONFIG"); + if (driverConfig != null) { + driverConfigs.add(driverConfig); + } + } + + // DRIVER_CONFIG is stored for exactly one connection (the control connection). Stage 1 reports + // only the schema version. + assertThat(driverConfigs).hasSize(1); + assertThat(driverConfigs.get(0)).isEqualTo("{\"version\":1}"); + } + + @Test(groups = "short") + public void should_share_session_id_across_multiple_sessions_from_same_cluster() { + // The Cluster-scoped session id, as generated by the driver and reported by the class-level + // session's connections. + String expectedSessionId = sessionId(cluster()); + assertThat(awaitClusterConnections(expectedSessionId, 2)) + .as("initial connections carrying this cluster's SESSION_ID") + .isNotEmpty(); + + // A second Session from the SAME Cluster reuses that Cluster's single session id (see + // Connection.Factory), so its connections must carry the identical SESSION_ID rather than a + // fresh one. + Set existingConnections = connectionKeys(allDriverRows()); + try (Session secondSession = cluster().connect()) { + List newRows = awaitNewDriverConnections(existingConnections, 1); + assertThat(newRows).as("new driver connections from the second session").isNotEmpty(); + for (Row row : newRows) { + assertThat(clientOptions(row).get("SESSION_ID")).isEqualTo(expectedSessionId); + } + } + } + + @Test(groups = "short") + public void should_store_session_id_but_no_driver_config_when_reporting_disabled() { + // The class-level session stays connected for the whole test class, so its connections would + // otherwise be indistinguishable from the disabled cluster's by driver_name alone. Snapshot the + // connections that already exist, and only look at ones that appear after. + Set existingConnections = connectionKeys(allDriverRows()); + + try (Cluster cluster = + register(createClusterBuilder().withDriverConfigReporting(false).build())) { + try (Session ignored = cluster.connect()) { + List rows = awaitNewDriverConnections(existingConnections, 2); + assertThat(rows.size()) + .as("new driver connections observed in system.clients") + .isGreaterThanOrEqualTo(2); + // SESSION_ID is sent regardless of the setting; only the config blob is suppressed. + String sessionId = sessionId(cluster); + for (Row row : rows) { + assertThat(clientOptions(row)) + .containsEntry("SESSION_ID", sessionId) + .doesNotContainKey("DRIVER_CONFIG"); + } + } + } + } + + /** + * The {@code SESSION_ID} the given (already initialized) cluster reports, read from the driver + * side so that a cluster's connections can be told apart from any other cluster's in {@code + * system.clients}. + */ + private String sessionId(Cluster cluster) { + return cluster.manager.connectionFactory.sessionId.toString(); + } + + /** + * Polls {@code system.clients} until at least {@code min} connections carrying the given {@code + * SESSION_ID} appear. + */ + private List awaitClusterConnections(String sessionId, int min) { + long deadline = System.currentTimeMillis() + 60_000L; + List rows = clusterConnections(sessionId); + while (rows.size() < min && System.currentTimeMillis() < deadline) { + try { + Thread.sleep(500L); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + rows = clusterConnections(sessionId); + } + return rows; + } + + /** + * Polls {@code system.clients} until at least {@code min} connections outside of {@code + * excludeKeys} appear, i.e. new connections opened after the snapshot was taken. + */ + private List awaitNewDriverConnections(Set excludeKeys, int min) { + long deadline = System.currentTimeMillis() + 60_000L; + List rows = newDriverConnections(excludeKeys); + while (rows.size() < min && System.currentTimeMillis() < deadline) { + try { + Thread.sleep(500L); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + rows = newDriverConnections(excludeKeys); + } + return rows; + } + + private List newDriverConnections(Set excludeKeys) { + List rows = new ArrayList(); + for (Row row : allDriverRows()) { + if (!excludeKeys.contains(connectionKey(row))) { + rows.add(row); + } + } + return rows; + } + + /** + * The rows in {@code system.clients} for the connections carrying the given {@code SESSION_ID}. + */ + private List clusterConnections(String sessionId) { + List rows = new ArrayList(); + for (Row row : allDriverRows()) { + if (sessionId.equals(clientOptions(row).get("SESSION_ID"))) { + rows.add(row); + } + } + return rows; + } + + /** All rows in {@code system.clients} belonging to this driver, regardless of which session. */ + private List allDriverRows() { + ResultSet result = + session().execute("SELECT driver_name, address, port, client_options FROM system.clients"); + List rows = new ArrayList(); + for (Row row : result) { + if (DRIVER_NAME.equals(row.getString("driver_name"))) { + rows.add(row); + } + } + return rows; + } + + /** The client-side {@code (address, port)} identifying each of the given connections. */ + private Set connectionKeys(List rows) { + Set keys = new HashSet(); + for (Row row : rows) { + keys.add(connectionKey(row)); + } + return keys; + } + + private String connectionKey(Row row) { + return row.getObject("address") + ":" + row.getObject("port"); + } + + private Map clientOptions(Row row) { + Map options = row.getMap("client_options", String.class, String.class); + return options == null ? Collections.emptyMap() : options; + } +} diff --git a/pom.xml b/pom.xml index 6147628b73e..8b857f6edc3 100644 --- a/pom.xml +++ b/pom.xml @@ -219,12 +219,6 @@ ${jackson.version} - - com.fasterxml.jackson.dataformat - jackson-dataformat-yaml - ${jackson.version} - - org.glassfish javax.json