diff --git a/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java b/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java index 344678c8149..6a4de17ad0e 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java +++ b/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java @@ -125,9 +125,22 @@ public class CCMBridge implements CCMAccess { *

At times it is necessary to use a separate java install for CCM then what is being used for * running tests. For example, if you want to run tests with JDK 6 but against Cassandra 2.0, * which requires JDK 7. + * + *

This is the environment for the globally configured server version, assembled by {@link + * #buildGlobalEnvironmentMap}; a cluster that configures its own version gets one derived from + * {@link #BASE_ENVIRONMENT_MAP} instead, see {@link + * Builder#buildEnvironmentMap(Builder.ResolvedVersions)}. */ private static final Map ENVIRONMENT_MAP; + /** + * {@link #ENVIRONMENT_MAP} without the variables that depend on which server flavor and version + * is installed, i.e. everything a cluster configuring its own version can reuse. {@code + * SCYLLA_PRODUCT} is stripped even if the surrounding process defined it, so that it can only + * ever be re-added for a version that was actually resolved as Enterprise. + */ + private static final Map BASE_ENVIRONMENT_MAP; + /** * A mapping of full DSE versions to their C* counterpart. This is not meant to be comprehensive. * If C* version cannot be derived, the method makes a 'best guess'. @@ -200,6 +213,8 @@ public class CCMBridge implements CCMAccess { String branch = System.getProperty("cassandra.branch"); // Inherit the current environment. Map envMap = Maps.newHashMap(new ProcessBuilder().environment()); + boolean globalScyllaEnterprise = false; + boolean globalScyllaBranchSpec = false; ImmutableSet.Builder installArgs = ImmutableSet.builder(); if (installDirectory != null && !installDirectory.trim().isEmpty()) { installArgs.add("--install-dir=" + new File(installDirectory).getAbsolutePath()); @@ -211,12 +226,9 @@ public class CCMBridge implements CCMAccess { installArgs.add("-v release:" + inputScyllaVersion); } else { installArgs.add("-v " + inputScyllaVersion); + globalScyllaBranchSpec = true; } - // Detect Scylla Enterprise - it should start with - // a 4-digit year. - if (inputScyllaVersion.matches("\\d{4}\\..*")) { - envMap.put("SCYLLA_PRODUCT", "enterprise"); - } + globalScyllaEnterprise = isScyllaEnterpriseVersion(inputScyllaVersion); } else if (inputCassandraVersion != null && !inputCassandraVersion.trim().isEmpty()) { installArgs.add("-v " + inputCassandraVersion); } @@ -248,7 +260,13 @@ public class CCMBridge implements CCMAccess { if (ccmJavaHome != null) { envMap.put("JAVA_HOME", ccmJavaHome); } - ENVIRONMENT_MAP = ImmutableMap.copyOf(envMap); + ENVIRONMENT_MAP = + buildGlobalEnvironmentMap(envMap, globalScyllaEnterprise, globalScyllaBranchSpec); + // A cluster that configures its own version derives SCYLLA_PRODUCT from that version instead, + // so an inherited value must not reach it: it would install an explicitly configured OSS + // version from the Enterprise repository. + envMap.remove("SCYLLA_PRODUCT"); + BASE_ENVIRONMENT_MAP = ImmutableMap.copyOf(envMap); if (isDse()) { GLOBAL_DSE_VERSION_NUMBER = VersionNumber.parse(inputCassandraVersion); @@ -348,6 +366,53 @@ public static boolean isWindows() { return osName != null && osName.startsWith("Windows"); } + /** + * Scylla Enterprise versions start with a 4-digit year (e.g. 2026.1.0), OSS ones don't. CCM + * installs them from a different repository, selected with the {@code SCYLLA_PRODUCT} variable. + */ + private static boolean isScyllaEnterpriseVersion(String versionString) { + return versionString != null && versionString.matches("\\d{4}\\..*"); + } + + /** + * Builds {@link #ENVIRONMENT_MAP}, the environment for the globally configured server version, + * from an inherited process environment. + * + *

Package-private and free of static state so that the {@code SCYLLA_PRODUCT} decision is + * assertable without a live process environment -- it is otherwise only reachable through a + * static initializer. + * + *

An inherited {@code SCYLLA_PRODUCT} is honoured only for a branch spec, where {@link + * #isScyllaEnterpriseVersion} cannot classify the string and exporting the variable is the only + * way to select the repository. For a version number, a pure Cassandra run, or no configured + * version, the resolved flavor wins and an inherited value is dropped: otherwise a stale {@code + * SCYLLA_PRODUCT=enterprise} in the surrounding shell would install an OSS version from the + * Enterprise repository. + */ + static Map buildGlobalEnvironmentMap( + Map inheritedEnvironment, + boolean scyllaEnterprise, + boolean scyllaBranchSpec) { + if (scyllaEnterprise) { + return withScyllaEnterprise(inheritedEnvironment); + } + if (scyllaBranchSpec) { + return ImmutableMap.copyOf(inheritedEnvironment); + } + Map envMap = Maps.newHashMap(inheritedEnvironment); + envMap.remove("SCYLLA_PRODUCT"); + return ImmutableMap.copyOf(envMap); + } + + /** Adds the CCM variable that makes Scylla install from the Enterprise repository. */ + private static Map withScyllaEnterprise(Map environmentMap) { + // Not an ImmutableMap.Builder: the inherited environment may already define the variable, and + // duplicate keys would make build() throw. + Map envMap = Maps.newHashMap(environmentMap); + envMap.put("SCYLLA_PRODUCT", "enterprise"); + return ImmutableMap.copyOf(envMap); + } + private static boolean isVersionNumber(String versionString) { try { VersionNumber.parse(versionString); @@ -408,6 +473,9 @@ private static VersionNumber parseScyllaInputVersion(String versionString) { private final int[] jmxPorts; + /** The environment to use for this cluster's CCM commands, see {@link #ENVIRONMENT_MAP}. */ + private final Map environmentMap; + protected CCMBridge( String clusterName, VersionNumber cassandraVersion, @@ -419,7 +487,8 @@ protected CCMBridge( int binaryPort, int[] jmxPorts, String jvmArgs, - int[] nodes) { + int[] nodes, + Map environmentMap) { this.clusterName = clusterName; this.cassandraVersion = cassandraVersion; @@ -430,11 +499,12 @@ protected CCMBridge( this.thriftPort = thriftPort; this.binaryPort = binaryPort; this.isDSE = dseVersion != null; - this.isScylla = (getGlobalScyllaVersion() != null); + this.isScylla = (scyllaVersion != null); this.jvmArgs = jvmArgs; this.nodes = nodes; this.ccmDir = Files.createTempDir(); this.jmxPorts = jmxPorts; + this.environmentMap = environmentMap; } public static Builder builder() { @@ -480,7 +550,7 @@ public InetSocketAddress addressOfNode(int n) { @Override public InetSocketAddress jmxAddressOfNode(int n) { - if (GLOBAL_SCYLLA_VERSION_NUMBER != null) { + if (isScylla) { return new InetSocketAddress(ipOfNode(n), jmxPorts[n - 1]); } else { return new InetSocketAddress("localhost", jmxPorts[n - 1]); @@ -733,24 +803,38 @@ public void add(int n) { public void add(int dc, int n) { logger.debug( String.format("Adding: node %s (%s%s:%s) to %s", n, ipPrefix, n, binaryPort, this)); - String thriftItf = ipOfNode(n) + ":" + thriftPort; String storageItf = ipOfNode(n) + ":" + storagePort; String binaryItf = ipOfNode(n) + ":" + binaryPort; String remoteLogItf = ipOfNode(n) + ":" + TestUtils.findAvailablePort(); - execute( - CCM_COMMAND - + " add node%d -d dc%s -i %s%d -t %s -l %s --binary-itf %s -j %d -r %s -s -b" - + (isDSE ? " --dse" : "") - + (isScylla ? " --scylla" : ""), - n, - dc, - ipPrefix, - n, - thriftItf, - storageItf, - binaryItf, - TestUtils.findAvailablePort(), - remoteLogItf); + if (isScylla) { + // scylla-ccm's `add` command has no thrift option: Scylla never had a Thrift interface. + execute( + CCM_COMMAND + + " add node%d -d dc%s -i %s%d -l %s --binary-itf %s -j %d -r %s -s -b --scylla", + n, + dc, + ipPrefix, + n, + storageItf, + binaryItf, + TestUtils.findAvailablePort(), + remoteLogItf); + } else { + String thriftItf = ipOfNode(n) + ":" + thriftPort; + execute( + CCM_COMMAND + + " add node%d -d dc%s -i %s%d -t %s -l %s --binary-itf %s -j %d -r %s -s -b" + + (isDSE ? " --dse" : ""), + n, + dc, + ipPrefix, + n, + thriftItf, + storageItf, + binaryItf, + TestUtils.findAvailablePort(), + remoteLogItf); + } } @Override @@ -837,7 +921,19 @@ private static VersionNumber getScyllaVersionThroughCcm(String versionString) { } } + /** + * Runs a CCM command with the environment of the globally configured server version. + * + *

Note that {@link #getScyllaVersionThroughCcm(String)} reaches this from the static + * initializer, before that environment is assigned; commons-exec then inherits this process's + * environment as-is. + */ private static String execute(File ccmDir, String command, Object... args) { + return execute(ccmDir, ENVIRONMENT_MAP, command, args); + } + + private static String execute( + File ccmDir, Map environmentMap, String command, Object... args) { Logger logger = CCMBridge.logger; String fullCommand = String.format(command, args) + " --config-dir=" + ccmDir; Closer closer = Closer.create(); @@ -873,7 +969,7 @@ protected void processLine(String line, int logLevel) { ExecuteStreamHandler streamHandler = new PumpStreamHandler(outStream, errStream); executor.setStreamHandler(streamHandler); executor.setWatchdog(watchDog); - int retValue = executor.execute(cli, ENVIRONMENT_MAP); + int retValue = executor.execute(cli, environmentMap); if (retValue != 0) { logger.error( "Non-zero exit code ({}) returned from executing ccm command: {}", @@ -903,7 +999,7 @@ protected void processLine(String line, int logLevel) { } private String execute(String command, Object... args) { - return execute(this.ccmDir, command, args); + return execute(this.ccmDir, this.environmentMap, command, args); } /** @@ -1005,6 +1101,8 @@ public static class Builder { private boolean start = true; private boolean dse = isDse(); private boolean scylla = GLOBAL_SCYLLA_VERSION_NUMBER != null; + private boolean ssl = false; + private boolean auth = false; private VersionNumber version = null; private final Set createOptions = new LinkedHashSet(); private final Set jvmArgs = new LinkedHashSet(); @@ -1046,40 +1144,22 @@ public Builder withClusterName(String clusterName) { return this; } - /** Enables SSL encryption. */ + /** + * Enables SSL encryption. + * + *

Only records the request: which keys and certificates to point the server at depends on + * the server flavor, which isn't resolved until {@link #build()}, see {@link + * #buildClientEncryptionOptions(ResolvedVersions)}. + */ public Builder withSSL() { - cassandraConfiguration.put("client_encryption_options.enabled", "true"); - if (GLOBAL_SCYLLA_VERSION_NUMBER != null) { - cassandraConfiguration.put( - "client_encryption_options.certificate", - DEFAULT_SERVER_CERT_CHAIN_FILE.getAbsolutePath()); - cassandraConfiguration.put( - "client_encryption_options.keyfile", DEFAULT_SERVER_PRIVATE_KEY_FILE.getAbsolutePath()); - } else { - cassandraConfiguration.put("client_encryption_options.optional", "false"); - cassandraConfiguration.put( - "client_encryption_options.keystore", DEFAULT_SERVER_KEYSTORE_FILE.getAbsolutePath()); - cassandraConfiguration.put( - "client_encryption_options.keystore_password", DEFAULT_SERVER_KEYSTORE_PASSWORD); - } + this.ssl = true; return this; } /** Enables client authentication. This also enables encryption ({@link #withSSL()}. */ public Builder withAuth() { withSSL(); - cassandraConfiguration.put("client_encryption_options.require_client_auth", "true"); - if (GLOBAL_SCYLLA_VERSION_NUMBER != null) { - cassandraConfiguration.put( - "client_encryption_options.truststore", - DEFAULT_SERVER_TRUSTSTORE_PEM_FILE.getAbsolutePath()); - } else { - cassandraConfiguration.put( - "client_encryption_options.truststore", - DEFAULT_SERVER_TRUSTSTORE_FILE.getAbsolutePath()); - cassandraConfiguration.put( - "client_encryption_options.truststore_password", DEFAULT_SERVER_TRUSTSTORE_PASSWORD); - } + this.auth = true; return this; } @@ -1092,6 +1172,12 @@ public Builder notStarted() { /** * The Cassandra or DSE or Scylla version to use. If not specified the globally configured * version is used instead. + * + *

Which of the three this version names is decided by {@link #withDSE(boolean)} and {@link + * #withScylla(boolean)}, which default to the flavor of the surrounding run rather than to + * anything about this version. Call the matching one alongside this method, or a Cassandra + * version passed under {@code -Dscylla.version=...} is resolved as a Scylla release (and vice + * versa). */ public Builder withVersion(VersionNumber version) { this.version = version; @@ -1172,37 +1258,60 @@ public Builder withWorkload(int node, Workload... workload) { return this; } - public CCMBridge build() { - // be careful NOT to alter internal state (hashCode/equals) during build! - String clusterName = TestUtils.generateIdentifier("ccm_"); - - if (providedClusterName != null) clusterName = providedClusterName; + /** The server versions this builder's configuration resolves to. */ + static class ResolvedVersions { + final boolean versionConfigured; + final VersionNumber cassandra; + final VersionNumber dse; + final VersionNumber scylla; + + ResolvedVersions( + boolean versionConfigured, + VersionNumber cassandra, + VersionNumber dse, + VersionNumber scylla) { + this.versionConfigured = versionConfigured; + this.cassandra = cassandra; + this.dse = dse; + this.scylla = scylla; + } + } - VersionNumber dseVersion; - VersionNumber cassandraVersion; - VersionNumber scyllaVersion; + /** + * Resolves which flavor and version this builder will create, from the explicitly configured + * version (if any) and the globally configured defaults. + */ + ResolvedVersions resolveVersions() { boolean versionConfigured = this.version != null; // No version was explicitly provided, fallback on global config. if (!versionConfigured) { - scyllaVersion = GLOBAL_SCYLLA_VERSION_NUMBER; - dseVersion = GLOBAL_DSE_VERSION_NUMBER; - cassandraVersion = GLOBAL_CASSANDRA_VERSION_NUMBER; + return new ResolvedVersions( + false, + GLOBAL_CASSANDRA_VERSION_NUMBER, + GLOBAL_DSE_VERSION_NUMBER, + GLOBAL_SCYLLA_VERSION_NUMBER); } else if (dse) { // given version is the DSE version, base cassandra version on DSE version. - scyllaVersion = null; - dseVersion = this.version; - cassandraVersion = getCassandraVersion(dseVersion); + return new ResolvedVersions(true, getCassandraVersion(this.version), this.version, null); } else if (scylla) { - scyllaVersion = this.version; - dseVersion = null; // Versions from 5.1 to 6.2.0 seem to report release_version 3.0.8 in system_local - cassandraVersion = VersionNumber.parse("3.0.8"); + return new ResolvedVersions(true, VersionNumber.parse("3.0.8"), null, this.version); } else { // given version is cassandra version. - scyllaVersion = null; - dseVersion = null; - cassandraVersion = this.version; + return new ResolvedVersions(true, this.version, null, null); } + } + + public CCMBridge build() { + // be careful NOT to alter internal state (hashCode/equals) during build! + String clusterName = TestUtils.generateIdentifier("ccm_"); + + if (providedClusterName != null) clusterName = providedClusterName; + + ResolvedVersions versions = resolveVersions(); + VersionNumber dseVersion = versions.dse; + VersionNumber cassandraVersion = versions.cassandra; + VersionNumber scyllaVersion = versions.scylla; Map cassandraConfiguration = randomizePorts(this.cassandraConfiguration); int storagePort = Integer.parseInt(cassandraConfiguration.get("storage_port").toString()); @@ -1241,12 +1350,19 @@ public CCMBridge build() { cassandraConfiguration.put("enable_sasi_indexes", true); } } - if (GLOBAL_SCYLLA_VERSION_NUMBER != null) { + if (scyllaVersion != null) { cassandraConfiguration.put("prometheus_port", RANDOM_PORT); cassandraConfiguration.put("api_port", RANDOM_PORT); cassandraConfiguration.put("native_shard_aware_transport_port", RANDOM_PORT); cassandraConfiguration = randomizePorts(cassandraConfiguration); } + // Explicitly configured entries keep winning over these defaults, as they did when withSSL() + // wrote them at builder-configuration time and withCassandraConfiguration() ran after it. + for (Map.Entry option : buildClientEncryptionOptions(versions).entrySet()) { + if (!cassandraConfiguration.containsKey(option.getKey())) { + cassandraConfiguration.put(option.getKey(), option.getValue()); + } + } final CCMBridge ccm = new CCMBridge( clusterName, @@ -1259,7 +1375,8 @@ public CCMBridge build() { binaryPort, generatedJmxPorts, joinJvmArgs(), - nodes); + nodes, + buildEnvironmentMap(versions)); Runtime.getRuntime() .addShutdownHook( @@ -1269,7 +1386,7 @@ public void run() { ccm.close(); } }); - ccm.execute(buildCreateCommand(clusterName, versionConfigured, cassandraVersion, dseVersion)); + ccm.execute(buildCreateCommand(clusterName, versions)); updateNodeConf(ccm); ccm.updateConfig(cassandraConfiguration); if (dseVersion != null) { @@ -1333,11 +1450,7 @@ private String joinJvmArgs() { return allJvmArgs.toString(); } - private String buildCreateCommand( - String clusterName, - boolean versionConfigured, - VersionNumber cassandraVersion, - VersionNumber dseVersion) { + String buildCreateCommand(String clusterName, ResolvedVersions versions) { StringBuilder result = new StringBuilder(CCM_COMMAND + " create"); result.append(" ").append(clusterName); result.append(" -i ").append(ipPrefix); @@ -1352,23 +1465,92 @@ private String buildCreateCommand( } Set lCreateOptions = new LinkedHashSet(createOptions); - if (!versionConfigured) { + if (!versions.versionConfigured) { // If no version was provided, use the default install ags. lCreateOptions.addAll(CASSANDRA_INSTALL_ARGS); } else { - if (dseVersion != null) { + if (versions.dse != null) { lCreateOptions.add("--dse"); lCreateOptions.add("-v"); - lCreateOptions.add(dseVersion.toString()); + lCreateOptions.add(versions.dse.toString()); + } else if (versions.scylla != null) { + // Same shape as the Scylla entries of CASSANDRA_INSTALL_ARGS. Which repository this + // installs from is decided by buildEnvironmentMap. + lCreateOptions.add("--scylla"); + lCreateOptions.add("-v"); + lCreateOptions.add("release:" + versions.scylla); } else { lCreateOptions.add("-v"); - lCreateOptions.add(cassandraVersion.toString()); + lCreateOptions.add(versions.cassandra.toString()); } } result.append(" ").append(Joiner.on(" ").join(randomizePorts(lCreateOptions))); return result.toString(); } + /** + * The client encryption yaml for a cluster with these versions, empty unless {@link #withSSL()} + * or {@link #withAuth()} was called. + * + *

Cassandra and DSE read a JKS keystore and truststore, Scylla reads a PEM certificate, key + * and truststore, so the flavor has to be resolved first — which is why this can't be decided + * in {@code withSSL()} itself. + */ + Map buildClientEncryptionOptions(ResolvedVersions versions) { + if (!ssl) { + return ImmutableMap.of(); + } + boolean isScylla = versions.scylla != null; + Map options = Maps.newLinkedHashMap(); + options.put("client_encryption_options.enabled", "true"); + if (isScylla) { + options.put( + "client_encryption_options.certificate", + DEFAULT_SERVER_CERT_CHAIN_FILE.getAbsolutePath()); + options.put( + "client_encryption_options.keyfile", DEFAULT_SERVER_PRIVATE_KEY_FILE.getAbsolutePath()); + } else { + options.put("client_encryption_options.optional", "false"); + options.put( + "client_encryption_options.keystore", DEFAULT_SERVER_KEYSTORE_FILE.getAbsolutePath()); + options.put( + "client_encryption_options.keystore_password", DEFAULT_SERVER_KEYSTORE_PASSWORD); + } + if (auth) { + options.put("client_encryption_options.require_client_auth", "true"); + if (isScylla) { + options.put( + "client_encryption_options.truststore", + DEFAULT_SERVER_TRUSTSTORE_PEM_FILE.getAbsolutePath()); + } else { + options.put( + "client_encryption_options.truststore", + DEFAULT_SERVER_TRUSTSTORE_FILE.getAbsolutePath()); + options.put( + "client_encryption_options.truststore_password", DEFAULT_SERVER_TRUSTSTORE_PASSWORD); + } + } + return options; + } + + /** + * The environment for the CCM commands of a cluster with these versions. + * + *

{@code SCYLLA_PRODUCT} has to follow the version this particular cluster installs, not the + * globally configured one: otherwise an explicitly configured Enterprise version would install + * from the OSS repository (and vice-versa). + */ + static Map buildEnvironmentMap(ResolvedVersions versions) { + if (!versions.versionConfigured) { + // The global environment is already derived from the same version. + return ENVIRONMENT_MAP; + } else if (versions.scylla != null && isScyllaEnterpriseVersion(versions.scylla.toString())) { + return withScyllaEnterprise(BASE_ENVIRONMENT_MAP); + } else { + return BASE_ENVIRONMENT_MAP; + } + } + /** * This is a workaround for an oddity in CCM: when we create a cluster with -n option and * non-standard ports, the node.conf files are not updated accordingly. @@ -1464,6 +1646,10 @@ public boolean equals(Object o) { if (ipPrefix != builder.ipPrefix) return false; if (dse != builder.dse) return false; if (scylla != builder.scylla) return false; + // Not reflected in cassandraConfiguration until build(), so they have to be compared here — + // otherwise an encrypted cluster could be reused for a test that expects a plaintext one. + if (ssl != builder.ssl) return false; + if (auth != builder.auth) return false; if (!Arrays.equals(nodes, builder.nodes)) return false; if (version != null ? !version.equals(builder.version) : builder.version != null) return false; @@ -1479,6 +1665,9 @@ public int hashCode() { // do not include start as it is not relevant to the settings of the cluster. int result = Arrays.hashCode(nodes); result = 31 * result + (dse ? 1 : 0); + result = 31 * result + (scylla ? 1 : 0); + result = 31 * result + (ssl ? 1 : 0); + result = 31 * result + (auth ? 1 : 0); result = 31 * result + ipPrefix.hashCode(); result = 31 * result + (version != null ? version.hashCode() : 0); result = 31 * result + createOptions.hashCode(); diff --git a/driver-core/src/test/java/com/datastax/driver/core/CCMBridgeCreateCommandTest.java b/driver-core/src/test/java/com/datastax/driver/core/CCMBridgeCreateCommandTest.java new file mode 100644 index 00000000000..3a8fbcb07bc --- /dev/null +++ b/driver-core/src/test/java/com/datastax/driver/core/CCMBridgeCreateCommandTest.java @@ -0,0 +1,256 @@ +package com.datastax.driver.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.datastax.driver.core.CCMBridge.Builder.ResolvedVersions; +import com.google.common.collect.ImmutableMap; +import java.util.Map; +import org.testng.annotations.Test; + +/** + * Unit tests for the part of {@link CCMBridge.Builder} that decides which server flavor and version + * to install, i.e. the {@code ccm create} command, the environment it runs in and the + * flavor-specific yaml it writes. No CCM cluster is created. + * + *

Each test configures the flavor explicitly, so that it doesn't depend on the {@code + * scylla.version} / {@code dse} system properties of the surrounding run. + */ +public class CCMBridgeCreateCommandTest { + + @Test(groups = "unit") + public void should_create_scylla_cluster_when_scylla_version_configured() { + CCMBridge.Builder builder = + CCMBridge.builder() + .withDSE(false) + .withScylla(true) + .withVersion(VersionNumber.parse("2026.1.0")); + + ResolvedVersions versions = builder.resolveVersions(); + assertThat(versions.scylla).isEqualTo(VersionNumber.parse("2026.1.0")); + assertThat(versions.cassandra).isEqualTo(VersionNumber.parse("3.0.8")); + assertThat(versions.dse).isNull(); + + String command = builder.buildCreateCommand("test_cluster", versions); + assertThat(command).contains("--scylla").contains("-v release:2026.1.0"); + assertThat(command).doesNotContain("--dse"); + // 3.0.8 is only what Scylla reports in system.local, it is never an install target + assertThat(command).doesNotContain("3.0.8"); + + // 2026.1.0 is an Enterprise version, it must not be installed from the OSS repository + assertThat(CCMBridge.Builder.buildEnvironmentMap(versions)) + .containsEntry("SCYLLA_PRODUCT", "enterprise"); + } + + @Test(groups = "unit") + public void should_not_use_enterprise_repository_for_open_source_scylla_version() { + CCMBridge.Builder builder = + CCMBridge.builder() + .withDSE(false) + .withScylla(true) + .withVersion(VersionNumber.parse("6.2.0")); + + ResolvedVersions versions = builder.resolveVersions(); + assertThat(versions.scylla).isEqualTo(VersionNumber.parse("6.2.0")); + + // Would leak in from a `-Dscylla.version=.` run if the product was global + assertThat(CCMBridge.Builder.buildEnvironmentMap(versions)).doesNotContainKey("SCYLLA_PRODUCT"); + } + + @Test(groups = "unit") + public void should_create_cassandra_cluster_when_cassandra_version_configured() { + CCMBridge.Builder builder = + CCMBridge.builder() + .withDSE(false) + .withScylla(false) + .withVersion(VersionNumber.parse("4.1.3")); + + ResolvedVersions versions = builder.resolveVersions(); + assertThat(versions.cassandra).isEqualTo(VersionNumber.parse("4.1.3")); + assertThat(versions.dse).isNull(); + assertThat(versions.scylla).isNull(); + + String command = builder.buildCreateCommand("test_cluster", versions); + assertThat(command).contains("-v 4.1.3"); + assertThat(command).doesNotContain("--scylla").doesNotContain("--dse"); + + assertThat(CCMBridge.Builder.buildEnvironmentMap(versions)).doesNotContainKey("SCYLLA_PRODUCT"); + } + + /** + * The globally configured version keeps the environment built for it in the static initializer: + * that one is derived from the raw {@code scylla.version} string, which may be a branch spec + * whose resolved version number looks like an Enterprise one without being installed as such. + */ + @Test(groups = "unit") + public void should_use_global_environment_when_no_version_configured() { + VersionNumber cassandra = VersionNumber.parse("3.0.8"); + VersionNumber enterpriseScylla = VersionNumber.parse("2026.1.0"); + + assertThat( + CCMBridge.Builder.buildEnvironmentMap( + new ResolvedVersions(false, cassandra, null, enterpriseScylla))) + .isSameAs( + CCMBridge.Builder.buildEnvironmentMap( + new ResolvedVersions(false, cassandra, null, null))); + } + + /** + * Scylla reads a PEM certificate and key, not the JKS keystore Cassandra reads, so an explicitly + * configured Scylla cluster must not be given the Cassandra settings just because the surrounding + * run has no {@code scylla.version}. + */ + @Test(groups = "unit") + public void should_use_pem_client_encryption_for_configured_scylla_version() { + CCMBridge.Builder builder = + CCMBridge.builder() + .withDSE(false) + .withScylla(true) + .withVersion(VersionNumber.parse("2026.1.0")) + .withAuth(); + + Map options = builder.buildClientEncryptionOptions(builder.resolveVersions()); + + assertThat(options) + .containsEntry("client_encryption_options.enabled", "true") + .containsEntry("client_encryption_options.require_client_auth", "true") + .containsKey("client_encryption_options.certificate") + .containsKey("client_encryption_options.keyfile") + .containsKey("client_encryption_options.truststore"); + assertThat(options) + .doesNotContainKey("client_encryption_options.keystore") + .doesNotContainKey("client_encryption_options.keystore_password") + .doesNotContainKey("client_encryption_options.truststore_password"); + } + + /** The mirror image: an explicit Cassandra version under a global Scylla run. */ + @Test(groups = "unit") + public void should_use_keystore_client_encryption_for_configured_cassandra_version() { + CCMBridge.Builder builder = + CCMBridge.builder() + .withDSE(false) + .withScylla(false) + .withVersion(VersionNumber.parse("4.1.3")) + .withAuth(); + + Map options = builder.buildClientEncryptionOptions(builder.resolveVersions()); + + assertThat(options) + .containsEntry("client_encryption_options.enabled", "true") + .containsEntry("client_encryption_options.require_client_auth", "true") + .containsKey("client_encryption_options.keystore") + .containsKey("client_encryption_options.keystore_password") + .containsKey("client_encryption_options.truststore") + .containsKey("client_encryption_options.truststore_password"); + assertThat(options) + .doesNotContainKey("client_encryption_options.certificate") + .doesNotContainKey("client_encryption_options.keyfile"); + } + + /** {@code withSSL()} alone must not enable client certificate authentication. */ + @Test(groups = "unit") + public void should_not_require_client_auth_without_with_auth() { + CCMBridge.Builder sslOnly = CCMBridge.builder().withDSE(false).withScylla(true).withSSL(); + assertThat(sslOnly.buildClientEncryptionOptions(sslOnly.resolveVersions())) + .containsEntry("client_encryption_options.enabled", "true") + .doesNotContainKey("client_encryption_options.require_client_auth"); + + CCMBridge.Builder plaintext = CCMBridge.builder().withDSE(false).withScylla(true); + assertThat(plaintext.buildClientEncryptionOptions(plaintext.resolveVersions())).isEmpty(); + } + + /** + * {@code ssl}/{@code auth} are no longer reflected in {@code cassandraConfiguration} at + * configuration time, so {@link CCMBridge.Builder} has to compare them itself: {@link CCMCache} + * keys cached clusters on the builder, and would otherwise hand an encrypted cluster to a test + * that asked for a plaintext one. + */ + @Test(groups = "unit") + public void should_not_consider_encrypted_and_plaintext_clusters_equal() { + CCMBridge.Builder plaintext = CCMBridge.builder().withNodes(1); + CCMBridge.Builder encrypted = CCMBridge.builder().withNodes(1).withSSL(); + CCMBridge.Builder authenticated = CCMBridge.builder().withNodes(1).withAuth(); + + assertThat(plaintext).isNotEqualTo(encrypted).isNotEqualTo(authenticated); + assertThat(encrypted).isNotEqualTo(authenticated); + assertThat(encrypted).isEqualTo(CCMBridge.builder().withNodes(1).withSSL()); + assertThat(encrypted.hashCode()) + .isEqualTo(CCMBridge.builder().withNodes(1).withSSL().hashCode()); + } + + @Test(groups = "unit") + public void should_create_dse_cluster_when_dse_version_configured() { + CCMBridge.Builder builder = + CCMBridge.builder() + .withDSE(true) + .withScylla(false) + .withVersion(VersionNumber.parse("6.8.0")); + + ResolvedVersions versions = builder.resolveVersions(); + assertThat(versions.dse).isEqualTo(VersionNumber.parse("6.8.0")); + assertThat(versions.cassandra).isNotNull(); + assertThat(versions.scylla).isNull(); + + String command = builder.buildCreateCommand("test_cluster", versions); + assertThat(command).contains("--dse").contains("-v 6.8.0"); + assertThat(command).doesNotContain("--scylla"); + } + + /** + * An environment as inherited from a shell that exported {@code SCYLLA_PRODUCT}, e.g. left over + * from an earlier step of the same CI job. + */ + private static Map inheritedEnterpriseEnvironment() { + return ImmutableMap.of("PATH", "/usr/bin", "SCYLLA_PRODUCT", "enterprise"); + } + + @Test(groups = "unit") + public void should_use_enterprise_repository_for_global_enterprise_version() { + Map environment = + CCMBridge.buildGlobalEnvironmentMap(inheritedEnterpriseEnvironment(), true, false); + + assertThat(environment).containsEntry("SCYLLA_PRODUCT", "enterprise"); + assertThat(environment).containsEntry("PATH", "/usr/bin"); + } + + /** + * The global version is a number that isn't Enterprise, so the repository is known: an inherited + * value must not override it, or an OSS version is looked up in the Enterprise repository. + */ + @Test(groups = "unit") + public void should_drop_inherited_product_for_global_open_source_version() { + Map environment = + CCMBridge.buildGlobalEnvironmentMap(inheritedEnterpriseEnvironment(), false, false); + + assertThat(environment).doesNotContainKey("SCYLLA_PRODUCT"); + assertThat(environment).containsEntry("PATH", "/usr/bin"); + } + + /** + * A branch spec can't be classified as Enterprise or OSS by its version string, so exporting + * {@code SCYLLA_PRODUCT} is the only way to select the repository: that one inherited value has + * to survive. + */ + @Test(groups = "unit") + public void should_keep_inherited_product_for_global_branch_spec() { + Map environment = + CCMBridge.buildGlobalEnvironmentMap(inheritedEnterpriseEnvironment(), false, true); + + assertThat(environment).containsEntry("SCYLLA_PRODUCT", "enterprise"); + } + + /** + * A pure Cassandra run, or no configured version at all: nothing about the run asks for the + * Enterprise repository, so a stale inherited value must not reach ccm. + */ + @Test(groups = "unit") + public void should_drop_inherited_product_when_no_scylla_version_configured() { + Map environment = + CCMBridge.buildGlobalEnvironmentMap( + ImmutableMap.of("JAVA_HOME", "/opt/java", "SCYLLA_PRODUCT", "enterprise"), + false, + false); + + assertThat(environment).doesNotContainKey("SCYLLA_PRODUCT"); + assertThat(environment).containsEntry("JAVA_HOME", "/opt/java"); + } +} diff --git a/driver-core/src/test/java/com/datastax/driver/core/CCMConfig.java b/driver-core/src/test/java/com/datastax/driver/core/CCMConfig.java index d43b39db95b..4f21a8494b8 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/CCMConfig.java +++ b/driver-core/src/test/java/com/datastax/driver/core/CCMConfig.java @@ -50,15 +50,20 @@ final class Undefined {} int[] numberOfNodes() default {}; /** - * The C* or DSE version to use; defaults to the version defined by the System property {@code - * cassandra.version}. + * The C*, DSE or Scylla version to use; defaults to the version defined by the System property + * {@code cassandra.version}. * *

Note that setting this attribute completely overrides the System properties {@code * cassandra.version} and {@code cassandra.directory}. * + *

Which server this version names is decided by {@link #dse()} and {@link #scylla()}, which + * default to the flavor of the surrounding run. Set the matching one explicitly whenever this + * attribute is set, or a Cassandra version will be installed as Scylla (or vice versa) depending + * on how the test run was invoked. + * *

This attribute is ignored if {@link #ccmProvider()} is defined. * - * @return The C* or DSE version to use + * @return The C*, DSE or Scylla version to use * @see CCMBridge#getCassandraVersion() */ String version() default ""; @@ -75,6 +80,20 @@ final class Undefined {} */ boolean[] dse() default {}; + /** + * Whether to launch a Scylla instance rather than an OSS C*. + * + *

Note that setting this attribute completely overrides the System property {@code + * scylla.version}: only whether Scylla is launched, not which version. Set it together with + * {@link #version()} so that an explicitly configured version is installed as the server it + * actually names, instead of inheriting the flavor of the surrounding run. + * + *

This attribute is ignored if {@link #ccmProvider()} is defined. + * + * @return {@code true} to launch a Scylla instance, {@code false} to launch an OSS C* instance. + */ + boolean[] scylla() default {}; + /** * Configuration items to add to cassandra.yaml configuration file. Each configuration item must * be in the form {@code key:value}. diff --git a/driver-core/src/test/java/com/datastax/driver/core/CCMTestsSupport.java b/driver-core/src/test/java/com/datastax/driver/core/CCMTestsSupport.java index c8627f1d0b4..1af9c7695cc 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/CCMTestsSupport.java +++ b/driver-core/src/test/java/com/datastax/driver/core/CCMTestsSupport.java @@ -344,6 +344,14 @@ private Boolean dse() { return null; } + @SuppressWarnings("SimplifiableIfStatement") + private Boolean scylla() { + for (CCMConfig ann : annotations) { + if (ann != null && ann.scylla().length > 0) return ann.scylla()[0]; + } + return null; + } + @SuppressWarnings("SimplifiableIfStatement") private boolean ssl() { for (CCMConfig ann : annotations) { @@ -497,14 +505,19 @@ private CCMBridge.Builder ccmBuilder(Object testInstance) throws Exception { ccmBuilder = CCMBridge.builder().withNodes(numberOfNodes()).notStarted(); } + // Set the flavor before the version: which server an explicitly configured version names + // is decided by these flags, which otherwise default to the flavor of the surrounding run. + Boolean dse = dse(); + if (dse != null) ccmBuilder.withDSE(dse); + Boolean scylla = scylla(); + if (scylla != null) ccmBuilder.withScylla(scylla); + String versionStr = version(); if (versionStr != null) { VersionNumber version = VersionNumber.parse(versionStr); ccmBuilder.withVersion(version); } - Boolean dse = dse(); - if (dse != null) ccmBuilder.withDSE(dse); if (ssl()) ccmBuilder.withSSL(); if (auth()) ccmBuilder.withAuth(); for (Map.Entry entry : config().entrySet()) { diff --git a/driver-core/src/test/java/com/datastax/driver/core/ProtocolVersionRenegotiationTest.java b/driver-core/src/test/java/com/datastax/driver/core/ProtocolVersionRenegotiationTest.java index 5de374b66d7..9552ca2b5b5 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/ProtocolVersionRenegotiationTest.java +++ b/driver-core/src/test/java/com/datastax/driver/core/ProtocolVersionRenegotiationTest.java @@ -84,7 +84,7 @@ public void should_fail_when_beta_allowed_and_too_high() { /** @jira_ticket JAVA-1367 */ @Test(groups = "short", enabled = false /* @IntegrationTestDisabledCassandra3Failure */) - @CCMConfig(version = "2.1.16", createCluster = false) + @CCMConfig(version = "2.1.16", scylla = false, createCluster = false) public void should_negotiate_when_no_version_provided() { if (protocolVersion.compareTo(ProtocolVersion.NEWEST_SUPPORTED) >= 0) { throw new SkipException("Server supports newest protocol version driver supports"); diff --git a/driver-core/src/test/java/com/datastax/driver/core/RecommissionedNodeTest.java b/driver-core/src/test/java/com/datastax/driver/core/RecommissionedNodeTest.java index 3e9907c57aa..e3686eedf88 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/RecommissionedNodeTest.java +++ b/driver-core/src/test/java/com/datastax/driver/core/RecommissionedNodeTest.java @@ -172,6 +172,8 @@ public void should_ignore_node_that_does_not_support_protocol_version_on_session .withStoragePort(mainCcm.getStoragePort()) .withThriftPort(mainCcm.getThriftPort()) .withBinaryPort(mainCcm.getBinaryPort()) + // 2.1.20 is a Cassandra version: say so, or a Scylla run would install it as Scylla. + .withScylla(false) .withVersion(VersionNumber.parse("2.1.20")); otherCcm = CCMCache.get(otherCcmBuilder); otherCcm.waitForUp(1); diff --git a/driver-core/src/test/java/com/datastax/driver/core/TabletsIT.java b/driver-core/src/test/java/com/datastax/driver/core/TabletsTest.java similarity index 93% rename from driver-core/src/test/java/com/datastax/driver/core/TabletsIT.java rename to driver-core/src/test/java/com/datastax/driver/core/TabletsTest.java index f3ecdd362c0..4b4694ed2fe 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/TabletsIT.java +++ b/driver-core/src/test/java/com/datastax/driver/core/TabletsTest.java @@ -24,8 +24,8 @@ }) @ScyllaOnly @ScyllaVersion(minOSS = "6.0.0", minEnterprise = "2024.2", description = "Needs to support tablets") -public class TabletsIT extends CCMTestsSupport { - private static final Logger LOG = LoggerFactory.getLogger(TabletsIT.class); +public class TabletsTest extends CCMTestsSupport { + private static final Logger LOG = LoggerFactory.getLogger(TabletsTest.class); private static final int INITIAL_TABLETS = 32; private static final int QUERIES = 1600; private static final int REPLICATION_FACTOR = 2; @@ -190,8 +190,13 @@ public void every_statement_should_deliver_tablet_info() { continue; } Session session = sessionEntry.getValue().get(); - // Empty out tablets information - session.getCluster().getMetadata().getTabletMap().removeTableMappings(KEYSPACE_NAME); + // Empty out tablets information. The mapping is keyed by the lowercased keyspace name, as + // reported by the server, so the key has to be lowercased here too or this is a no-op. + session + .getCluster() + .getMetadata() + .getTabletMap() + .removeTableMappings(KEYSPACE_NAME.toLowerCase()); Statement stmt; try { stmt = stmtEntry.getValue().apply(session); @@ -226,6 +231,10 @@ public void every_statement_should_deliver_tablet_info() { stmtEntry.getKey(), sessionEntry.getKey())); continue; } + // executeOnAllHostsAndReturnIfResultHasTabletsInfo pins the statement to a specific host + // while hunting for tablet info. Clear that pin, otherwise the routing check below always + // observes the pinned host and can never detect misrouting. + stmt.setHost(null); if (!checkIfRoutedProperly(session, stmt)) { testErrors.add( String.format( @@ -343,6 +352,11 @@ private static boolean checkIfRoutedProperly(Session session, Statement stmt) { int expectedNodesCount = stmt.isLWT() ? 1 : REPLICATION_FACTOR; Set nodes = new HashSet<>(); for (int i = 0; i < REPLICATION_FACTOR * 3; i++) { + // PagingOptimizingLoadBalancingPolicy returns Statement.getLastHost() ahead of the real query + // plan, and that field is set after every successful BoundStatement execution. Clearing it + // keeps the loop from being pinned to the first coordinator, which would let any routing + // behaviour satisfy the check below. + stmt.setLastHost(null); nodes.add(session.execute(stmt).getExecutionInfo().getQueriedHost()); } return nodes.size() <= expectedNodesCount; diff --git a/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesIT.java b/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesTest.java similarity index 96% rename from driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesIT.java rename to driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesTest.java index ebcf5bdf352..26c0041303e 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesIT.java +++ b/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesTest.java @@ -14,7 +14,7 @@ import org.testng.annotations.DataProvider; import org.testng.annotations.Test; -public class ZeroTokenNodesIT { +public class ZeroTokenNodesTest { @DataProvider(name = "loadBalancingPolicies") public static Object[][] loadBalancingPolicies() { @@ -173,12 +173,17 @@ public void should_discover_zero_token_DC_when_option_is_enabled( queriedNodes.add(rs.getExecutionInfo().getQueriedHost().getEndPoint().resolve()); } + // containsOnly, not containsExactly: queriedNodes is a HashSet, whose iteration order is + // hash-derived rather than insertion order, so an order-sensitive assertion is a latent + // flake. + // AssertJ 1.7.1, pinned by this module, has no containsExactlyInAnyOrder; for a Set, + // containsOnly is equivalent to it. if (isDcAware) { assertThat(queriedNodes) - .containsExactly(ccmBridge.addressOfNode(1), ccmBridge.addressOfNode(2)); + .containsOnly(ccmBridge.addressOfNode(1), ccmBridge.addressOfNode(2)); } else { assertThat(queriedNodes) - .containsExactly( + .containsOnly( ccmBridge.addressOfNode(1), ccmBridge.addressOfNode(2), ccmBridge.addressOfNode(3), diff --git a/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingIT.java b/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingIT.java deleted file mode 100644 index eed462e1dc6..00000000000 --- a/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingIT.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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.driver.core.policies; - -import static org.assertj.core.api.Assertions.assertThat; - -import com.datastax.driver.core.BoundStatement; -import com.datastax.driver.core.CCMConfig; -import com.datastax.driver.core.CCMTestsSupport; -import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.ConsistencyLevel; -import com.datastax.driver.core.Host; -import com.datastax.driver.core.PreparedStatement; -import com.datastax.driver.core.ResultSet; -import com.datastax.driver.core.Session; -import com.datastax.driver.core.SimpleStatement; -import java.net.InetSocketAddress; -import java.util.HashSet; -import java.util.Set; -import org.testng.annotations.Test; - -/** - * Integration tests verifying that statements with SERIAL/LOCAL_SERIAL consistency level are routed - * through the LWT load-balancing path (PRESERVE_REPLICA_ORDER). - */ -@CCMConfig(numberOfNodes = 3) -public class LWTLoadBalancingIT extends CCMTestsSupport { - - @Override - public Cluster.Builder createClusterBuilder() { - return Cluster.builder() - .withLoadBalancingPolicy( - new TokenAwarePolicy(new RoundRobinPolicy(), TokenAwarePolicy.ReplicaOrdering.RANDOM)); - } - - @Override - public void onTestContextInitialized() { - execute("CREATE TABLE IF NOT EXISTS test_lwt (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); - for (int i = 0; i < 10; i++) { - execute(String.format("INSERT INTO test_lwt (pk, ck, v) VALUES (%d, %d, %d)", i, 0, i)); - } - } - - @Test(groups = "short") - public void should_route_local_serial_select_through_lwt_path() { - Session session = session(); - - SimpleStatement simpleSelect = - new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?", 1, 0); - simpleSelect.setConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL); - - PreparedStatement preparedSelect = session.prepare(simpleSelect); - BoundStatement boundSelect = preparedSelect.bind(1, 0); - - // Verify statement properties - assertThat(simpleSelect.isLWT()).isFalse(); - assertThat(simpleSelect.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_SERIAL); - - // Execute multiple times and collect coordinators — with PRESERVE_REPLICA_ORDER routing, - // the same partition key should always be routed to the same first replica. - Set coordinators = new HashSet<>(); - for (int i = 0; i < 30; i++) { - ResultSet rs = session.execute(boundSelect); - Host coordinator = rs.getExecutionInfo().getQueriedHost(); - assertThat(coordinator).isNotNull(); - coordinators.add(coordinator.getEndPoint().resolve()); - } - - // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key, - // so all 30 executions should hit the same coordinator. - assertThat(coordinators).hasSize(1); - } - - @Test(groups = "short") - public void should_route_serial_select_through_lwt_path() { - Session session = session(); - - SimpleStatement simpleSelect = - new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?", 2, 0); - simpleSelect.setConsistencyLevel(ConsistencyLevel.SERIAL); - - PreparedStatement preparedSelect = session.prepare(simpleSelect); - BoundStatement boundSelect = preparedSelect.bind(2, 0); - - // Execute multiple times and collect coordinators - Set coordinators = new HashSet<>(); - for (int i = 0; i < 30; i++) { - ResultSet rs = session.execute(boundSelect); - Host coordinator = rs.getExecutionInfo().getQueriedHost(); - assertThat(coordinator).isNotNull(); - coordinators.add(coordinator.getEndPoint().resolve()); - } - - // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key. - assertThat(coordinators).hasSize(1); - } -} diff --git a/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingTest.java b/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingTest.java new file mode 100644 index 00000000000..f9eb78bc544 --- /dev/null +++ b/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingTest.java @@ -0,0 +1,188 @@ +/* + * 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.driver.core.policies; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.datastax.driver.core.BoundStatement; +import com.datastax.driver.core.CCMConfig; +import com.datastax.driver.core.CCMTestsSupport; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.ConsistencyLevel; +import com.datastax.driver.core.Host; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SimpleStatement; +import com.datastax.driver.core.TestUtils; +import com.google.common.base.Throwables; +import java.net.InetSocketAddress; +import java.util.HashSet; +import java.util.Objects; +import java.util.Set; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.annotations.Test; + +/** + * Integration tests verifying that statements with SERIAL/LOCAL_SERIAL consistency level are routed + * through the LWT load-balancing path (PRESERVE_REPLICA_ORDER). + */ +@CCMConfig(numberOfNodes = 3) +public class LWTLoadBalancingTest extends CCMTestsSupport { + + private static final Logger LOGGER = LoggerFactory.getLogger(LWTLoadBalancingTest.class); + + /** Equal to the node count, so that every node is a replica of every partition. */ + private static final int REPLICATION_FACTOR = 3; + + private static final int EXECUTIONS = 30; + + @Override + public Cluster.Builder createClusterBuilder() { + return Cluster.builder() + .withLoadBalancingPolicy( + new TokenAwarePolicy(new RoundRobinPolicy(), TokenAwarePolicy.ReplicaOrdering.RANDOM)); + } + + /** + * Override to create the keyspace with a replication factor greater than 1. The default test + * keyspace created by {@link CCMTestsSupport} is hardcoded to RF=1, and with a single replica per + * partition "the first replica" is trivially unique — every assertion below would hold under + * {@code REGULAR} routing too, so the tests could not tell {@code PRESERVE_REPLICA_ORDER} apart + * from {@code RANDOM}. + * + *

Tablets are disabled when running against Scylla: with tablets enabled, replica placement + * comes from the tablet map, which is empty until it has been learned from a misrouted query, and + * an empty replica list makes the LWT query plan fall back to the child policy — a non-replica + * coordinator on the first execution. Cassandra does not support the tablets property. + */ + @Override + protected void initTestKeyspace() { + try { + keyspace = TestUtils.generateIdentifier("ks_"); + LOGGER.debug("Using keyspace " + keyspace); + boolean isScylla = Objects.nonNull(ccm().getScyllaVersion()); + session() + .execute( + String.format( + "CREATE KEYSPACE %s WITH replication = {'class': 'NetworkTopologyStrategy'," + + " 'datacenter1': %d}" + + (isScylla ? " AND tablets = {'enabled': false}" : ""), + keyspace, + REPLICATION_FACTOR)); + useKeyspace(keyspace); + } catch (Exception e) { + errorOut(); + LOGGER.error("Could not create test keyspace", e); + Throwables.propagate(e); + } + } + + @Override + public void onTestContextInitialized() { + execute("CREATE TABLE IF NOT EXISTS test_lwt (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + for (int i = 0; i < 10; i++) { + execute(String.format("INSERT INTO test_lwt (pk, ck, v) VALUES (%d, %d, %d)", i, 0, i)); + } + } + + @Test(groups = "short") + public void should_route_local_serial_select_through_lwt_path() { + Session session = session(); + + SimpleStatement simpleSelect = + new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?"); + simpleSelect.setConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL); + + PreparedStatement preparedSelect = session.prepare(simpleSelect); + + // Verify statement properties + assertThat(simpleSelect.isLWT()).isFalse(); + assertThat(simpleSelect.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_SERIAL); + + // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key, + // so every execution should hit the same coordinator. Contrast with the non-serial control in + // should_spread_non_serial_select_across_replicas, which shares the same statement and policy. + assertThat(collectCoordinators(session, preparedSelect, 1)).hasSize(1); + } + + @Test(groups = "short") + public void should_route_serial_select_through_lwt_path() { + Session session = session(); + + SimpleStatement simpleSelect = + new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?"); + simpleSelect.setConsistencyLevel(ConsistencyLevel.SERIAL); + + PreparedStatement preparedSelect = session.prepare(simpleSelect); + + // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key. + assertThat(collectCoordinators(session, preparedSelect, 2)).hasSize(1); + } + + /** + * Control for the two tests above. This is the same statement against the same table, executed by + * the same {@code TokenAwarePolicy(RoundRobinPolicy, RANDOM)} — only the consistency level + * differs. A non-serial level takes the {@code REGULAR} routing path, which shuffles the replicas + * on every query, so the coordinator must vary. If this test ever collapses to a single + * coordinator as well, the {@code hasSize(1)} assertions above have stopped proving anything + * about {@code PRESERVE_REPLICA_ORDER}. + */ + @Test(groups = "short") + public void should_spread_non_serial_select_across_replicas() { + Session session = session(); + + SimpleStatement simpleSelect = + new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?"); + simpleSelect.setConsistencyLevel(ConsistencyLevel.ONE); + + PreparedStatement preparedSelect = session.prepare(simpleSelect); + BoundStatement boundSelect = preparedSelect.bind(3, 0); + + assertThat(boundSelect.isLWT()).isFalse(); + assertThat(boundSelect.getConsistencyLevel().isSerial()).isFalse(); + + // Uniform over REPLICATION_FACTOR replicas across EXECUTIONS queries, so the probability of a + // false failure here is REPLICATION_FACTOR^(1 - EXECUTIONS). + assertThat(collectCoordinators(session, preparedSelect, 3).size()).isGreaterThan(1); + } + + /** + * Executes {@code prepared} against partition {@code pk} {@link #EXECUTIONS} times and returns + * the distinct coordinators used. + * + *

A fresh {@link BoundStatement} is bound for every execution on purpose. {@link + * PagingOptimizingLoadBalancingPolicy}, which the driver wraps around the configured policy, + * returns {@code Statement.getLastHost()} ahead of the real query plan, and that field is set on + * every successful {@code BoundStatement} execution. Reusing a single instance would therefore + * pin the coordinator after the first query and make every assertion in this class hold + * regardless of how routing actually behaves. + */ + private static Set collectCoordinators( + Session session, PreparedStatement prepared, int pk) { + Set coordinators = new HashSet<>(); + for (int i = 0; i < EXECUTIONS; i++) { + ResultSet rs = session.execute(prepared.bind(pk, 0)); + Host coordinator = rs.getExecutionInfo().getQueriedHost(); + assertThat(coordinator).isNotNull(); + coordinators.add(coordinator.getEndPoint().resolve()); + } + return coordinators; + } +} diff --git a/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderIT.java b/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderTest.java similarity index 99% rename from driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderIT.java rename to driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderTest.java index 1a32eb2d88d..9e1b696ad4e 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderIT.java +++ b/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderTest.java @@ -37,7 +37,7 @@ import java.util.Iterator; import org.testng.annotations.Test; -public class SchemaBuilderIT extends CCMTestsSupport { +public class SchemaBuilderTest extends CCMTestsSupport { // Test relies on existence of 'ks' keyspace, // but no such keyspace is created. If (fixed) created,