You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by fr...@apache.org on 2024/04/04 23:15:54 UTC
(cassandra-analytics) branch trunk updated: CASSANDRA-19528: Use a classloader to isolate in-jvm dtest classes in… (#50)
This is an automated email from the ASF dual-hosted git repository.
frankgh pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new 47fdb64 CASSANDRA-19528: Use a classloader to isolate in-jvm dtest classes in… (#50)
47fdb64 is described below
commit 47fdb6448b6956249790d5dc7bb76b699d35c079
Author: Francisco Guerrero <fr...@apache.org>
AuthorDate: Thu Apr 4 16:15:49 2024 -0700
CASSANDRA-19528: Use a classloader to isolate in-jvm dtest classes in… (#50)
Patch by Francisco Guerrero; Reviewed by Yifan Cai for CASSANDRA-19528
---
.../spark/validation/StartupValidator.java | 2 +-
.../spark/validation/StartupValidatorTests.java | 12 +
.../build.gradle | 117 +++----
.../distributed/impl/CassandraCluster.java | 339 +++++++++++++++++++++
.../testing/JvmDTestSharedClassesPredicate.java | 47 ---
.../cassandra/sidecar/testing/QualifiedName.java | 16 +-
.../testing/SharedClusterIntegrationTestBase.java | 232 +++++++++-----
.../testing/ClusterBuilderConfiguration.java | 139 +++++++++
.../cassandra/testing/IClusterExtension.java | 102 +++++++
.../testing/IsolatedDTestClassLoaderWrapper.java | 210 +++++++++++++
.../cassandra/testing}/TestTokenSupplier.java | 10 +-
.../org/apache/cassandra/testing/TestUtils.java | 31 ++
.../cassandra/testing/utils/ClusterUtils.java | 183 +++++++++++
.../cassandra/testing/utils/WithProperties.java | 56 ++++
cassandra-analytics-integration-tests/build.gradle | 172 ++++++-----
.../cassandra/analytics/BlockedInstancesTest.java | 35 +--
.../cassandra/analytics/BulkWriteTtlTest.java | 37 +--
.../cassandra/analytics/BulkWriteUdtTest.java | 46 +--
.../analytics/QuoteIdentifiersReadTest.java | 57 +---
.../analytics/QuoteIdentifiersWriteTest.java | 68 ++---
.../analytics/ReadDifferentTablesTest.java | 30 --
.../cassandra/analytics/ResiliencyTestBase.java | 211 +------------
.../SharedClusterSparkIntegrationTestBase.java | 83 +++--
.../analytics/SparkBulkWriterSimpleTest.java | 30 +-
.../apache/cassandra/analytics/SparkTestUtils.java | 89 ++++--
.../analytics/SparkTestUtilsProvider.java | 56 ++++
.../analytics/TimestampIntegrationTest.java | 36 +--
.../analytics/WriterDigestIntegrationTest.java | 36 +--
.../analytics/data/ClearSnapshotTest.java | 43 +--
.../expansion/JoiningDoubleClusterFailureTest.java | 3 +-
.../expansion/JoiningDoubleClusterTest.java | 3 +-
.../expansion/JoiningMultiDCFailureTest.java | 3 +-
.../JoiningMultiDCSingleReplicatedFailureTest.java | 3 +-
.../JoiningMultiDCSingleReplicatedTest.java | 3 +-
.../analytics/expansion/JoiningMultiDCTest.java | 3 +-
.../expansion/JoiningMultipleNodesFailureTest.java | 3 +-
.../expansion/JoiningMultipleNodesTest.java | 3 +-
.../expansion/JoiningSingleNodeFailureTest.java | 3 +-
.../analytics/expansion/JoiningSingleNodeTest.java | 3 +-
.../analytics/expansion/JoiningTestBase.java | 44 ++-
.../movement/NodeMovementFailureTest.java | 3 +-
.../movement/NodeMovementMultiDCFailureTest.java | 3 +-
.../movement/NodeMovementMultiDCTest.java | 3 +-
.../analytics/movement/NodeMovementTest.java | 3 +-
.../analytics/movement/NodeMovementTestBase.java | 32 +-
.../replacement/HostReplacementFailureTest.java | 3 +-
.../HostReplacementMultiDCFailureTest.java | 3 +-
...ReplacementMultiDCInsufficientReplicasTest.java | 3 +-
.../replacement/HostReplacementMultiDCTest.java | 3 +-
.../analytics/replacement/HostReplacementTest.java | 3 +-
.../replacement/HostReplacementTestBase.java | 51 ++--
.../analytics/shrink/LeavingHalfTest.java | 3 +-
.../shrink/LeavingMultiDCFailureTest.java | 3 +-
.../LeavingMultiDCHalveClusterFailureTest.java | 3 +-
.../shrink/LeavingMultiDCHalveClusterTest.java | 3 +-
.../analytics/shrink/LeavingMultiDCTest.java | 3 +-
.../shrink/LeavingMultipleFailureTest.java | 3 +-
.../analytics/shrink/LeavingMultipleTest.java | 3 +-
.../analytics/shrink/LeavingSingleFailureTest.java | 3 +-
.../analytics/shrink/LeavingSingleTest.java | 3 +-
.../analytics/shrink/LeavingTestBase.java | 26 +-
gradle/wrapper/gradle-wrapper.jar | Bin 59536 -> 63721 bytes
gradle/wrapper/gradle-wrapper.properties | 4 +-
gradlew | 31 +-
scripts/build-sidecar.sh | 2 +-
65 files changed, 1772 insertions(+), 1027 deletions(-)
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/validation/StartupValidator.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/validation/StartupValidator.java
index 9d8ed3c..7048dec 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/validation/StartupValidator.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/validation/StartupValidator.java
@@ -105,6 +105,6 @@ public final class StartupValidator
public boolean enabled()
{
- return System.getProperty(DISABLE) == null;
+ return !"true".equalsIgnoreCase(System.getProperty(DISABLE));
}
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/validation/StartupValidatorTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/validation/StartupValidatorTests.java
index 8fafce0..b610d5d 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/validation/StartupValidatorTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/validation/StartupValidatorTests.java
@@ -65,6 +65,18 @@ public class StartupValidatorTests
assertEquals("Failed some of startup validations", exception.getMessage());
}
+ @Test
+ public void testSkipValidations()
+ {
+ System.setProperty("SKIP_STARTUP_VALIDATIONS", "true");
+ StartupValidator.instance().register(TestValidation.succeeding());
+ StartupValidator.instance().register(TestValidation.failing());
+ StartupValidator.instance().register(TestValidation.succeeding());
+
+ StartupValidator.instance().perform();
+ System.clearProperty("SKIP_STARTUP_VALIDATIONS");
+ }
+
@AfterAll
public static void afterAll()
{
diff --git a/cassandra-analytics-integration-framework/build.gradle b/cassandra-analytics-integration-framework/build.gradle
index 7cc0fef..aeba617 100644
--- a/cassandra-analytics-integration-framework/build.gradle
+++ b/cassandra-analytics-integration-framework/build.gradle
@@ -17,76 +17,77 @@
* under the License.
*/
-project(':cassandra-analytics-integration-framework') {
- apply(plugin: 'java-library')
- apply(plugin: 'com.github.johnrengelman.shadow')
-
- ext.dtestJar = System.getenv("DTEST_JAR") ?: "dtest-4.1.4.jar" // latest supported Cassandra build is 4.1
- def dtestJarFullPath = "${dependencyLocation}${ext.dtestJar}"
+plugins {
+ id('java-library')
+ id('maven-publish')
+}
- test {
- useJUnitPlatform()
- }
+java {
+ withJavadocJar()
+ withSourcesJar()
+}
- configurations {
- configureEach {
- resolutionStrategy {
- // Force set the vertx versions that do not depend on a newer jackson version
- force("io.vertx:vertx-web:${vertxVersion}")
- force("io.vertx:vertx-core:${vertxVersion}")
- force("io.vertx:vertx-auth-common:${vertxVersion}")
- force("io.vertx:vertx-dropwizard-metrics:${vertxVersion}")
- force("io.vertx:vertx-web-client:${vertxVersion}")
- }
+publishing {
+ publications {
+ maven(MavenPublication) {
+ from components.java
+ groupId project.group
+ artifactId "${archivesBaseName}"
+ version System.getenv("CODE_VERSION") ?: "${version}"
}
}
+}
- dependencies {
+ext.dtestJar = System.getenv("DTEST_JAR") ?: "dtest-4.1.4.jar" // latest supported Cassandra build is 4.1
+def dtestJarFullPath = "${dependencyLocation}${ext.dtestJar}"
- // These dependencies (sidecar + dtest jars) must be built by running `scripts/build-dependencies.sh`
- implementation(group: 'org.apache.cassandra.sidecar', name: 'common', version: "${sidecarVersion}")
- implementation(group: 'org.apache.cassandra.sidecar', name: 'adapters-base', version: "${sidecarVersion}")
- // The server itself
- implementation (group: "org.apache.cassandra", name: "cassandra-sidecar", version: "${sidecarVersion}") {
- exclude(group: 'org.apache.logging.log4j')
- exclude(group: 'org.slf4j')
- exclude(group: 'ch.qos.logback')
- }
+test {
+ useJUnitPlatform()
+}
- File dtestJarFile = new File(dtestJarFullPath)
- assert dtestJarFile.exists()
- println("Using DTest jar: " + dtestJarFullPath)
- shadow(files("${dtestJarFullPath}"))
- shadow("org.apache.cassandra:dtest-api:0.0.16")
- // Needed by the Cassandra dtest framework
- // JUnit
- implementation("org.junit.jupiter:junit-jupiter-api:${project.junitVersion}")
- implementation("org.junit.jupiter:junit-jupiter-params:${project.junitVersion}")
- implementation("org.junit.jupiter:junit-jupiter-engine:${project.junitVersion}")
+dependencies {
+ // These dependencies (sidecar + dtest jars) must be built by running `scripts/build-dependencies.sh`
+ implementation(group: 'org.apache.cassandra.sidecar', name: 'adapters-base', version: "${sidecarVersion}")
+ implementation(group: 'org.apache.cassandra.sidecar', name: 'adapters-cassandra41', version: "${sidecarVersion}")
- implementation("org.junit.vintage:junit-vintage-engine:${junitVersion}")
- implementation('org.mockito:mockito-inline:4.10.0')
- implementation("org.assertj:assertj-core:3.24.2")
+ File dtestJarFile = new File(dtestJarFullPath)
+ println("Using DTest jar: " + dtestJarFullPath + "; present? " + dtestJarFile.exists())
+ // The dtest jar must be a compile only dependency. We want to avoid having this jar in the
+ // classpath while running integration tests. Instead, a dedicated classloader will load the
+ // dtest jar while provisioning the in-jvm dtest Cassandra cluster
+ compileOnly(files("${dtestJarFullPath}"))
+ api("org.apache.cassandra:dtest-api:0.0.16")
+ // Needed by the Cassandra dtest framework
+ // JUnit
+ api("org.junit.jupiter:junit-jupiter-api:${project.junitVersion}")
+ api("org.junit.jupiter:junit-jupiter-params:${project.junitVersion}")
+ api("org.junit.jupiter:junit-jupiter-engine:${project.junitVersion}")
- shadow('com.datastax.cassandra:cassandra-driver-core:3.9.0')
+ api("org.junit.vintage:junit-vintage-engine:${junitVersion}")
+ api('org.mockito:mockito-inline:4.10.0')
+ api("org.assertj:assertj-core:3.24.2")
- shadow("com.google.guava:guava:27.0.1-jre")
- shadow('com.google.inject:guice:4.2.2')
- shadow("io.vertx:vertx-web:${project.vertxVersion}") {
- exclude group: 'junit', module: 'junit'
- }
- shadow("io.vertx:vertx-web-client:${project.vertxVersion}")
- shadow("io.vertx:vertx-junit5:${project.vertxVersion}")
+ api('com.datastax.cassandra:cassandra-driver-core:3.11.3')
+ implementation('com.google.inject:guice:4.2.2')
- testImplementation(platform('org.junit:junit-bom:5.9.2'))
- testImplementation('org.junit.jupiter:junit-jupiter')
- testImplementation("org.assertj:assertj-core:3.24.2")
+ implementation("com.google.guava:guava:27.0.1-jre")
+ api("io.vertx:vertx-web:${project.vertxVersion}") {
+ exclude group: 'junit', module: 'junit'
}
+ implementation("io.vertx:vertx-web-client:${project.vertxVersion}")
+ implementation group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: '2.14.3'
- shadowJar {
- zip64 true
- relocate('org.apache.cassandra.sidecar', 'o.a.c.analytics.sidecar.shaded.testing') {
- exclude '%regex[org.apache.cassandra.sidecar.testing.*]'
- }
+ testImplementation(platform("org.junit:junit-bom:${project.junitVersion}"))
+ testImplementation('org.junit.jupiter:junit-jupiter')
+ testImplementation("org.assertj:assertj-core:3.24.2")
+
+ // Exposed to the test project
+ api("io.vertx:vertx-junit5:${project.vertxVersion}")
+ // The server itself
+ api(group: "org.apache.cassandra", name: "cassandra-sidecar", version: "${sidecarVersion}") {
+ exclude(group: 'org.apache.logging.log4j')
+ exclude(group: 'org.slf4j')
+ exclude(group: 'ch.qos.logback')
}
+ api(group: 'org.apache.cassandra.sidecar', name: 'common', version: "${sidecarVersion}")
}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java
new file mode 100644
index 0000000..4d20c62
--- /dev/null
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java
@@ -0,0 +1,339 @@
+/*
+ * 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 org.apache.cassandra.distributed.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.Spliterator;
+import java.util.function.BiPredicate;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import com.google.common.base.Preconditions;
+
+import com.vdurmont.semver4j.Semver;
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.IMessageSink;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.shared.Versions;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
+import org.apache.cassandra.testing.IClusterExtension;
+import org.apache.cassandra.testing.TestTokenSupplier;
+import org.jetbrains.annotations.NotNull;
+
+import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack;
+import static org.apache.cassandra.distributed.shared.NetworkTopology.networkTopology;
+
+/**
+ * Implements the {@link IClusterExtension} interface, delegating the functionality to the
+ * {@link AbstractCluster delegate}. This class is meant to be loaded in a different classloader.
+ *
+ * @param <I> the type of the cluster instance
+ */
+public class CassandraCluster<I extends IInstance> implements IClusterExtension<I>
+{
+ private final AbstractCluster<I> delegate;
+
+ private static final Predicate<String> EXTRA = className -> {
+ // Those classes can be reached by Spark SizeEstimator, when it estimates the broadcast variable.
+ // In the test scenario containing cassandra instance shutdown, there is a chance that it pick the class
+ // that is loaded by the closed instance classloader, causing the following exception.
+ // java.lang.IllegalStateException: Can't load <CLASS>. Instance class loader is already closed.
+ return className.equals("org.apache.cassandra.utils.concurrent.Ref$OnLeak")
+ || className.startsWith("org.apache.cassandra.metrics.RestorableMeter")
+ || (className.startsWith("org.apache.cassandra.analytics.") && className.contains("BBHelper"));
+ };
+
+ public CassandraCluster(String versionString, ClusterBuilderConfiguration configuration) throws IOException
+ {
+ delegate = initializeCluster(versionString, configuration);
+ }
+
+ @SuppressWarnings("unchecked")
+ public AbstractCluster<I> initializeCluster(String versionString,
+ ClusterBuilderConfiguration configuration) throws IOException
+ {
+ // spin up a C* cluster using the in-jvm dtest
+ Versions versions = Versions.find();
+ Versions.Version requestedVersion = versions.getLatest(new Semver(versionString, Semver.SemverType.LOOSE));
+
+ int nodesPerDc = configuration.nodesPerDc;
+ int dcCount = configuration.dcCount;
+ int newNodesPerDc = configuration.newNodesPerDc;
+ Preconditions.checkArgument(newNodesPerDc >= 0, "newNodesPerDc cannot be a negative number");
+ int originalNodeCount = nodesPerDc * dcCount;
+ int finalNodeCount = dcCount * (nodesPerDc + newNodesPerDc);
+
+ TokenSupplier tokenSupplier = TestTokenSupplier.evenlyDistributedTokens(nodesPerDc, newNodesPerDc, dcCount, 1);
+
+ UpgradeableCluster.Builder clusterBuilder = UpgradeableCluster.build(originalNodeCount);
+ clusterBuilder.withVersion(requestedVersion)
+ .withDynamicPortAllocation(configuration.dynamicPortAllocation) // to allow parallel test runs
+ .withSharedClasses(EXTRA.or(clusterBuilder.getSharedClasses()))
+ .withDCs(dcCount)
+ .withDataDirCount(configuration.numDataDirsPerInstance)
+ .withConfig(config -> configuration.features.forEach(config::with))
+ .withTokenSupplier(tokenSupplier);
+
+
+ if (dcCount > 1)
+ {
+ clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount,
+ (nodeId) -> nodeId % 2 != 0 ?
+ dcAndRack("datacenter1", "rack1") :
+ dcAndRack("datacenter2", "rack2")));
+ }
+
+ if (configuration.instanceInitializer != null)
+ {
+ clusterBuilder.withInstanceInitializer(configuration.instanceInitializer);
+ }
+
+ UpgradeableCluster cluster = clusterBuilder.start();
+ if (cluster.size() > 1)
+ {
+ waitForHealthyRing(cluster);
+ fixDistributedSchemas((AbstractCluster<I>) cluster);
+ }
+ return (AbstractCluster<I>) cluster;
+ }
+
+ // IClusterExtension methods
+
+ @Override
+ public void schemaChangeIgnoringStoppedInstances(String query)
+ {
+ delegate.schemaChangeIgnoringStoppedInstances(query);
+ }
+
+ @Override
+ public I addInstance(String dc, String rack, Consumer<IInstanceConfig> fn)
+ {
+ return ClusterUtils.addInstance(delegate, dc, rack, fn);
+ }
+
+ @Override
+ public I getFirstRunningInstance()
+ {
+ return delegate.getFirstRunningInstance();
+ }
+
+ @Override
+ public IInstanceConfig newInstanceConfig()
+ {
+ return delegate.newInstanceConfig();
+ }
+
+ @Override
+ public IInstanceConfig createInstanceConfig(int nodeNum)
+ {
+ return delegate.createInstanceConfig(nodeNum);
+ }
+
+ @Override
+ public ICluster<I> delegate()
+ {
+ return delegate;
+ }
+
+ @Override
+ public void awaitRingState(IInstance instance, IInstance expectedInRing, String state)
+ {
+ ClusterUtils.awaitRingState(instance, expectedInRing, state);
+ }
+
+ @Override
+ public void awaitGossipStatus(IInstance instance, IInstance expectedInGossip, String targetStatus)
+ {
+ ClusterUtils.awaitGossipStatus(instance, expectedInGossip, targetStatus);
+ }
+
+ @Override
+ public void stopUnchecked(IInstance instance)
+ {
+ ClusterUtils.stopUnchecked(instance);
+ }
+
+ // ICluster methods
+
+ @Override
+ public void startup()
+ {
+ delegate.startup();
+ }
+
+ @Override
+ public I bootstrap(IInstanceConfig iInstanceConfig)
+ {
+ return delegate.bootstrap(iInstanceConfig);
+ }
+
+ @Override
+ public I get(int i)
+ {
+ return delegate.get(i);
+ }
+
+ @Override
+ public I get(InetSocketAddress inetSocketAddress)
+ {
+ return delegate.get(inetSocketAddress);
+ }
+
+ @Override
+ public ICoordinator coordinator(int i)
+ {
+ return delegate.coordinator(i);
+ }
+
+ @Override
+ public void schemaChange(String s)
+ {
+ delegate.schemaChange(s);
+ }
+
+ @Override
+ public void schemaChange(String s, int i)
+ {
+ delegate.schemaChange(s, i);
+ }
+
+ @Override
+ public int size()
+ {
+ return delegate.size();
+ }
+
+ @Override
+ public Stream<I> stream()
+ {
+ return delegate.stream();
+ }
+
+ @Override
+ public Stream<I> stream(String s)
+ {
+ return delegate.stream(s);
+ }
+
+ @Override
+ public Stream<I> stream(String s, String s1)
+ {
+ return delegate.stream(s, s1);
+ }
+
+ @NotNull
+ @Override
+ public Iterator<I> iterator()
+ {
+ return delegate.iterator();
+ }
+
+ @Override
+ public IMessageFilters filters()
+ {
+ return delegate.filters();
+ }
+
+ @Override
+ public void setMessageSink(IMessageSink messageSink)
+ {
+ delegate.setMessageSink(messageSink);
+ }
+
+ @Override
+ public void deliverMessage(InetSocketAddress to, IMessage msg)
+ {
+ delegate.deliverMessage(to, msg);
+ }
+
+ @Override
+ public void setUncaughtExceptionsFilter(BiPredicate<Integer, Throwable> ignoreThrowable)
+ {
+ delegate.setUncaughtExceptionsFilter(ignoreThrowable);
+ }
+
+ @Override
+ public void setUncaughtExceptionsFilter(Predicate<Throwable> ignoreThrowable)
+ {
+ delegate.setUncaughtExceptionsFilter(ignoreThrowable);
+ }
+
+ @Override
+ public void checkAndResetUncaughtExceptions()
+ {
+ delegate.checkAndResetUncaughtExceptions();
+ }
+
+ @Override
+ public void close() throws Exception
+ {
+ delegate.close();
+ }
+
+ @Override
+ public void forEach(Consumer<? super I> action)
+ {
+ delegate.forEach(action);
+ }
+
+ @Override
+ public Spliterator<I> spliterator()
+ {
+ return delegate.spliterator();
+ }
+
+ // Utility methods
+
+ void fixDistributedSchemas(AbstractCluster<I> cluster)
+ {
+ // These keyspaces are under replicated by default, so must be updated when doing a multi-node cluster;
+ // else bootstrap will fail with 'Unable to find sufficient sources for streaming range <range> in keyspace <name>'
+ for (String ks : Arrays.asList("system_auth", "system_traces"))
+ {
+ cluster.schemaChange("ALTER KEYSPACE " + ks +
+ " WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': "
+ + Math.min(cluster.size(), 3) + "}",
+ true,
+ cluster.getFirstRunningInstance());
+ }
+
+ // in real live repair is needed in this case, but in the test case it doesn't matter if the tables loose
+ // anything, so ignoring repair to speed up the tests.
+ }
+
+ void waitForHealthyRing(ICluster<?> cluster)
+ {
+ for (IInstance inst : cluster)
+ {
+ ClusterUtils.awaitRingHealthy(inst);
+ }
+ }
+}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/JvmDTestSharedClassesPredicate.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/JvmDTestSharedClassesPredicate.java
deleted file mode 100644
index d4772d0..0000000
--- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/JvmDTestSharedClassesPredicate.java
+++ /dev/null
@@ -1,47 +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 org.apache.cassandra.sidecar.testing;
-
-import java.util.function.Predicate;
-
-import org.apache.cassandra.distributed.impl.AbstractCluster;
-
-/**
- * Predicate to instruct the JVM DTest framework on the classes should be loaded by the shared classloader.
- */
-public class JvmDTestSharedClassesPredicate implements Predicate<String>
-{
- public static final JvmDTestSharedClassesPredicate INSTANCE = new JvmDTestSharedClassesPredicate();
-
- private static final Predicate<String> EXTRA = className -> {
- // Those classes can be reached by Spark SizeEstimator, when it estimates the broadcast variable.
- // In the test scenario containing cassandra instance shutdown, there is a chance that it pick the class
- // that is loaded by the closed instance classloader, causing the following exception.
- // java.lang.IllegalStateException: Can't load <CLASS>. Instance class loader is already closed.
- return className.equals("org.apache.cassandra.utils.concurrent.Ref$OnLeak")
- || className.startsWith("org.apache.cassandra.metrics.RestorableMeter");
- };
- private static final Predicate<String> DELEGATE = AbstractCluster.SHARED_PREDICATE.or(EXTRA);
-
- @Override
- public boolean test(String s)
- {
- return DELEGATE.test(s);
- }
-}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/QualifiedName.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/QualifiedName.java
index bf63423..accf05a 100644
--- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/QualifiedName.java
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/QualifiedName.java
@@ -20,8 +20,6 @@ package org.apache.cassandra.sidecar.testing;
import java.util.Objects;
-import org.apache.cassandra.cql3.ColumnIdentifier;
-
/**
* Simple class representing a Cassandra table, used for integration testing
*/
@@ -29,11 +27,21 @@ public class QualifiedName
{
private final String keyspace;
private final String table;
+ private final String maybeQuotedKeyspace;
+ private final String maybeQuotedTable;
public QualifiedName(String keyspace, String table)
+ {
+ this(keyspace, table, false, false);
+ }
+
+ public QualifiedName(String keyspace, String table,
+ boolean quoteKeyspace, boolean quoteTable)
{
this.keyspace = keyspace;
this.table = table;
+ this.maybeQuotedKeyspace = quoteKeyspace ? "\"" + keyspace + "\"" : keyspace;
+ this.maybeQuotedTable = quoteTable ? "\"" + table + "\"" : table;
}
public String keyspace()
@@ -43,7 +51,7 @@ public class QualifiedName
public String maybeQuotedKeyspace()
{
- return ColumnIdentifier.maybeQuote(keyspace);
+ return maybeQuotedKeyspace;
}
public String table()
@@ -53,7 +61,7 @@ public class QualifiedName
public String maybeQuotedTable()
{
- return ColumnIdentifier.maybeQuote(table);
+ return maybeQuotedTable;
}
@Override
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java
index a4e098d..60299ce 100644
--- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java
@@ -35,7 +35,6 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import org.apache.commons.lang3.StringUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestInstance;
@@ -43,6 +42,10 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SimpleStatement;
import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Injector;
@@ -52,15 +55,12 @@ import com.google.inject.util.Modules;
import io.vertx.core.Vertx;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.distributed.api.IInstanceConfig;
-import org.apache.cassandra.distributed.impl.AbstractCluster;
import org.apache.cassandra.distributed.shared.JMXUtil;
-import org.apache.cassandra.distributed.shared.ShutdownException;
import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
import org.apache.cassandra.sidecar.cluster.InstancesConfig;
import org.apache.cassandra.sidecar.cluster.InstancesConfigImpl;
@@ -76,17 +76,16 @@ import org.apache.cassandra.sidecar.config.ServiceConfiguration;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
import org.apache.cassandra.sidecar.config.yaml.SidecarConfigurationImpl;
+import org.apache.cassandra.sidecar.exceptions.ThrowableUtils;
import org.apache.cassandra.sidecar.server.MainModule;
import org.apache.cassandra.sidecar.server.Server;
import org.apache.cassandra.sidecar.utils.CassandraVersionProvider;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
+import org.apache.cassandra.testing.IClusterExtension;
+import org.apache.cassandra.testing.IsolatedDTestClassLoaderWrapper;
import org.apache.cassandra.testing.TestUtils;
import org.apache.cassandra.testing.TestVersion;
import org.apache.cassandra.testing.TestVersionSupplier;
-import org.apache.cassandra.utils.Throwables;
-import relocated.shaded.com.datastax.driver.core.Cluster;
-import relocated.shaded.com.datastax.driver.core.ResultSet;
-import relocated.shaded.com.datastax.driver.core.Session;
-import relocated.shaded.com.datastax.driver.core.SimpleStatement;
import static org.assertj.core.api.Assertions.assertThat;
@@ -102,7 +101,9 @@ import static org.assertj.core.api.Assertions.assertThat;
*
* <ol>
* <li>Find the first version from the {@link TestVersionSupplier#testVersions()}
+ * <li>(Optional) Before cluster provisioning (implementer can supply)
* <li>Provision a cluster for the test using the version from the previous step (implementer must supply)
+ * <li>(Optional) After cluster provisioned (implementer can supply)
* <li>Initialize schemas required for the test (implementer must supply)
* <li>Start sidecar that talks to the provisioned cluster
* <li>(Optional) Run the before test start method (implementer can supply)
@@ -132,10 +133,11 @@ public abstract class SharedClusterIntegrationTestBase
protected Vertx vertx;
protected DnsResolver dnsResolver;
- protected AbstractCluster<? extends IInstance> cluster;
+ protected IClusterExtension<? extends IInstance> cluster;
protected Server server;
protected Injector injector;
protected TestVersion testVersion;
+ private IsolatedDTestClassLoaderWrapper classLoaderWrapper;
static
{
@@ -144,32 +146,45 @@ public abstract class SharedClusterIntegrationTestBase
}
@BeforeAll
- protected void setup() throws InterruptedException, IOException
+ protected void setup() throws InterruptedException
{
Optional<TestVersion> maybeTestVersion = TestVersionSupplier.testVersions().findFirst();
assertThat(maybeTestVersion).isPresent();
this.testVersion = maybeTestVersion.get();
logger.info("Testing with version={}", testVersion);
+
+ classLoaderWrapper = new IsolatedDTestClassLoaderWrapper();
+ classLoaderWrapper.initializeDTestJarClassLoader(testVersion, SharedClusterIntegrationTestBase.class);
+
+ beforeClusterProvisioning();
cluster = provisionClusterWithRetries(this.testVersion);
assertThat(cluster).isNotNull();
+ afterClusterProvisioned();
initializeSchemaForTest();
startSidecar(cluster);
beforeTestStart();
}
- protected AbstractCluster<? extends IInstance> provisionClusterWithRetries(TestVersion testVersion) throws IOException
+ /**
+ * Provisions a cluster with the provided {@link TestVersion}. Up to {@link #MAX_CLUSTER_PROVISION_RETRIES}
+ * attempts will be made to provision a cluster when it fails to provision.
+ *
+ * @param testVersion the version for the test
+ * @return the provisioned cluster
+ */
+ private IClusterExtension<? extends IInstance> provisionClusterWithRetries(TestVersion testVersion)
{
for (int retry = 0; retry < MAX_CLUSTER_PROVISION_RETRIES; retry++)
{
try
{
- return provisionCluster(testVersion);
+ return classLoaderWrapper.loadCluster(testVersion.version(), testClusterConfiguration());
}
catch (RuntimeException runtimeException)
{
- boolean addressAlreadyInUse = Throwables.anyCauseMatches(runtimeException,
- ex -> ex instanceof BindException &&
- StringUtils.contains(ex.getMessage(), "Address already in use"));
+ boolean addressAlreadyInUse = ThrowableUtils.getCause(runtimeException, ex -> ex instanceof BindException &&
+ ex.getMessage() != null &&
+ ex.getMessage().contains("Address already in use")) != null;
if (addressAlreadyInUse)
{
logger.warn("Failed to provision cluster after {} retries", retry, runtimeException);
@@ -180,25 +195,57 @@ public abstract class SharedClusterIntegrationTestBase
}
}
}
- throw new RuntimeException("Unable to provision cluster");
+ throw new RuntimeException("Unable to provision cluster after " + MAX_CLUSTER_PROVISION_RETRIES + " retries");
}
@AfterAll
- protected void tearDown() throws InterruptedException
+ protected void tearDown() throws Exception
+ {
+ try
+ {
+ beforeSidecarStop();
+ stopSidecar();
+ beforeClusterShutdown();
+ closeCluster();
+ afterClusterShutdown();
+ }
+ finally
+ {
+ if (classLoaderWrapper != null)
+ {
+ classLoaderWrapper.closeDTestJarClassLoader();
+ }
+ }
+ }
+
+ /**
+ * Returns the configuration for the test cluster. The default configuration for the cluster has 1
+ * node, 1 DC, 1 data directory per node, with the {@link org.apache.cassandra.distributed.api.Feature#GOSSIP},
+ * {@link org.apache.cassandra.distributed.api.Feature#JMX}, and
+ * {@link org.apache.cassandra.distributed.api.Feature#NATIVE_PROTOCOL} features enabled. It uses dynamic port
+ * allocation for the Cassandra service ports. This method can be overridden to provide a different configuration
+ * for the cluster.
+ *
+ * @return the configuration for the test cluster
+ */
+ protected ClusterBuilderConfiguration testClusterConfiguration()
+ {
+ return new ClusterBuilderConfiguration();
+ }
+
+ /**
+ * Override to perform an action before the cluster provisioning
+ */
+ protected void beforeClusterProvisioning()
{
- beforeSidecarStop();
- stopSidecar();
- beforeClusterShutdown();
- closeCluster();
- afterClusterShutdown();
}
/**
- * @param testVersion the Cassandra version to use for the test
- * @return a provisioned cluster to use for tests
- * @throws IOException when provisioning a cluster fails
+ * Override to perform an action after the cluster has been successfully provisioned
*/
- protected abstract UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException;
+ protected void afterClusterProvisioned()
+ {
+ }
/**
* Initialize required schemas for the tests upfront before the test starts
@@ -256,10 +303,10 @@ public abstract class SharedClusterIntegrationTestBase
* @param cluster the cluster to use
* @throws InterruptedException when the startup times out
*/
- protected void startSidecar(AbstractCluster<? extends IInstance> cluster) throws InterruptedException
+ protected void startSidecar(ICluster<? extends IInstance> cluster) throws InterruptedException
{
VertxTestContext context = new VertxTestContext();
- injector = Guice.createInjector(Modules.override(new MainModule()).with(new IntegrationTestModule(cluster)));
+ injector = Guice.createInjector(Modules.override(new MainModule()).with(new IntegrationTestModule(cluster, classLoaderWrapper)));
dnsResolver = injector.getInstance(DnsResolver.class);
vertx = injector.getInstance(Vertx.class);
server = injector.getInstance(Server.class);
@@ -295,8 +342,10 @@ public abstract class SharedClusterIntegrationTestBase
/**
* Closes the cluster and its resources
+ *
+ * @throws Exception on an exception generated during cluster shutdown
*/
- protected void closeCluster()
+ protected void closeCluster() throws Exception
{
if (cluster == null)
{
@@ -311,7 +360,8 @@ public abstract class SharedClusterIntegrationTestBase
// `catch (ShutdownException)` won't always work - compare the canonical names instead.
catch (Throwable t)
{
- if (Objects.equals(t.getClass().getCanonicalName(), ShutdownException.class.getCanonicalName()))
+ if (Objects.equals(t.getClass().getCanonicalName(),
+ "org.apache.cassandra.distributed.shared.ShutdownException"))
{
logger.debug("Encountered shutdown exception which closing the cluster", t);
}
@@ -353,6 +403,17 @@ public abstract class SharedClusterIntegrationTestBase
return cluster.coordinator(1).execute(String.format("SELECT * FROM %s;", table), consistencyLevel);
}
+ /**
+ * Convenience method to query all data from the provided {@code table} at consistency level ALL.
+ *
+ * @param table the qualified Cassandra table name
+ * @return all the data queried from the table
+ */
+ protected ResultSet queryAllDataWithDriver(QualifiedName table)
+ {
+ return queryAllDataWithDriver(table, com.datastax.driver.core.ConsistencyLevel.ALL);
+ }
+
/**
* Convenience method to query all data from the provided {@code table} at the specified consistency level.
*
@@ -360,49 +421,47 @@ public abstract class SharedClusterIntegrationTestBase
* @param consistency the consistency level to use for querying the data
* @return all the data queried from the table
*/
- protected ResultSet queryAllDataWithDriver(ICluster<?> cluster, QualifiedName table,
- relocated.shaded.com.datastax.driver.core.ConsistencyLevel consistency)
+ protected ResultSet queryAllDataWithDriver(QualifiedName table,
+ com.datastax.driver.core.ConsistencyLevel consistency)
{
- Cluster driverCluster = createDriverCluster(cluster);
+ Cluster driverCluster = createDriverCluster(cluster.delegate());
Session session = driverCluster.connect();
SimpleStatement statement = new SimpleStatement(String.format("SELECT * FROM %s;", table));
statement.setConsistencyLevel(consistency);
return session.execute(statement);
}
+ // Utility methods
+
public static Cluster createDriverCluster(ICluster<? extends IInstance> dtest)
{
- if (dtest.size() == 0)
- {
- throw new IllegalArgumentException("Attempted to open java driver for empty cluster");
- }
- else
- {
- dtest.stream().forEach((i) -> {
- if (!i.config().has(Feature.NATIVE_PROTOCOL) || !i.config().has(Feature.GOSSIP))
- {
- throw new IllegalStateException("Java driver requires Feature.NATIVE_PROTOCOL and Feature.GOSSIP; " +
- "but one or more is missing");
- }
- });
- Cluster.Builder builder = Cluster.builder();
- dtest.stream().forEach((i) -> {
- InetSocketAddress address = new InetSocketAddress(i.broadcastAddress().getAddress(),
- i.config().getInt("native_transport_port"));
- builder.addContactPointsWithPorts(address);
- });
-
- return builder.build();
- }
+ dtest.stream().forEach((i) -> {
+ if (!i.config().has(Feature.NATIVE_PROTOCOL) || !i.config().has(Feature.GOSSIP))
+ {
+ throw new IllegalStateException("Java driver requires Feature.NATIVE_PROTOCOL and Feature.GOSSIP; " +
+ "but one or more is missing");
+ }
+ });
+ Cluster.Builder builder = Cluster.builder()
+ .withoutMetrics();
+ dtest.stream().forEach((i) -> {
+ InetSocketAddress address = new InetSocketAddress(i.broadcastAddress().getAddress(),
+ i.config().getInt("native_transport_port"));
+ builder.addContactPointsWithPorts(address);
+ });
+
+ return builder.build();
}
static class IntegrationTestModule extends AbstractModule
{
- private final AbstractCluster<? extends IInstance> cluster;
+ private final ICluster<? extends IInstance> cluster;
+ private final IsolatedDTestClassLoaderWrapper wrapper;
- IntegrationTestModule(AbstractCluster<? extends IInstance> cluster)
+ IntegrationTestModule(ICluster<? extends IInstance> cluster, IsolatedDTestClassLoaderWrapper wrapper)
{
this.cluster = cluster;
+ this.wrapper = wrapper;
}
@Provides
@@ -420,14 +479,15 @@ public abstract class SharedClusterIntegrationTestBase
SharedExecutorNettyOptions.INSTANCE);
List<InstanceMetadata> instanceMetadataList =
- IntStream.range(0, cluster.size())
+ IntStream.rangeClosed(1, cluster.size())
.mapToObj(i -> buildInstanceMetadata(vertx,
- cluster.get(i + 1),
+ cluster.get(i),
cassandraVersionProvider,
sidecarVersionProvider.sidecarVersion(),
jmxConfiguration,
cqlSessionProvider,
- dnsResolver))
+ dnsResolver,
+ wrapper))
.collect(Collectors.toList());
return new InstancesConfigImpl(instanceMetadataList, dnsResolver);
}
@@ -478,7 +538,8 @@ public abstract class SharedClusterIntegrationTestBase
String sidecarVersion,
JmxConfiguration jmxConfiguration,
CQLSessionProvider session,
- DnsResolver dnsResolver)
+ DnsResolver dnsResolver,
+ IsolatedDTestClassLoaderWrapper wrapper)
{
IInstanceConfig config = cassandraInstance.config();
String ipAddress = JMXUtil.getJmxHost(config);
@@ -495,12 +556,12 @@ public abstract class SharedClusterIntegrationTestBase
String[] dataDirectories = (String[]) config.get("data_file_directories");
String stagingDir = stagingDir(dataDirectories);
- JmxClient jmxClient = JmxClient.builder()
- .host(ipAddress)
- .port(config.jmxPort())
- .connectionMaxRetries(jmxConfiguration.maxRetries())
- .connectionRetryDelayMillis(jmxConfiguration.retryDelayMillis())
- .build();
+ JmxClient jmxClient = new JmxClientProxy(wrapper,
+ JmxClient.builder()
+ .host(ipAddress)
+ .port(config.jmxPort())
+ .connectionMaxRetries(jmxConfiguration.maxRetries())
+ .connectionRetryDelayMillis(jmxConfiguration.retryDelayMillis()));
CassandraAdapterDelegate delegate = new CassandraAdapterDelegate(vertx,
config.num(),
versionProvider,
@@ -531,4 +592,39 @@ public abstract class SharedClusterIntegrationTestBase
return stagingPath.toFile().getAbsolutePath();
}
}
+
+ protected JmxClient wrapJmxClient(JmxClient.Builder builder)
+ {
+ return new JmxClientProxy(classLoaderWrapper, builder);
+ }
+
+ /**
+ * Runs JMXClient calls with the dtest classloader
+ */
+ static class JmxClientProxy extends JmxClient
+ {
+ private final IsolatedDTestClassLoaderWrapper wrapper;
+
+ protected JmxClientProxy(IsolatedDTestClassLoaderWrapper wrapper, Builder builder)
+ {
+ super(Objects.requireNonNull(builder, "builder must be provided"));
+ this.wrapper = wrapper;
+ }
+
+ /**
+ * Connects to JMX by running inside the wrapped classloader. The connection to JMX must be established in the
+ * dtest classloader to be able to communicate to Cassandra's JMX.
+ *
+ * @param currentAttempt the current attempt to connect
+ * @throws IOException if the underlying operation throws an IOException
+ */
+ @Override
+ protected void connectInternal(int currentAttempt) throws IOException
+ {
+ wrapper.executeExceptionableActionOnDTestClassLoader(() -> {
+ super.connectInternal(currentAttempt);
+ return null;
+ });
+ }
+ }
}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/ClusterBuilderConfiguration.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/ClusterBuilderConfiguration.java
new file mode 100644
index 0000000..91ff516
--- /dev/null
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/ClusterBuilderConfiguration.java
@@ -0,0 +1,139 @@
+/*
+ * 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 org.apache.cassandra.testing;
+
+import java.util.EnumSet;
+import java.util.function.BiConsumer;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.distributed.api.Feature;
+
+/**
+ * Defines the configuration to build the {@link IClusterExtension} cluster
+ */
+public class ClusterBuilderConfiguration
+{
+ public int nodesPerDc = 1;
+ public int dcCount = 1;
+ public int newNodesPerDc = 0;
+ public int numDataDirsPerInstance = 1;
+ public boolean dynamicPortAllocation = true;
+ public final EnumSet<Feature> features = EnumSet.of(Feature.GOSSIP, Feature.JMX, Feature.NATIVE_PROTOCOL);
+ public BiConsumer<ClassLoader, Integer> instanceInitializer = null;
+
+ /**
+ * Adds a features to the list of default features.
+ *
+ * @param feature the {@code feature} to add
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration requestFeature(Feature feature)
+ {
+ features.add(feature);
+ return this;
+ }
+
+ /**
+ * Removes a feature to the list of requested features for the cluster.
+ *
+ * @param feature the {@code feature} to add
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration removeFeature(Feature feature)
+ {
+ features.remove(feature);
+ return this;
+ }
+
+ /**
+ * Sets the {@code nodesPerDc} and returns a reference to this Builder enabling method chaining.
+ *
+ * @param nodesPerDc the {@code nodesPerDc} to set
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration nodesPerDc(int nodesPerDc)
+ {
+ this.nodesPerDc = nodesPerDc;
+ return this;
+ }
+
+ /**
+ * Sets the {@code dcCount} and returns a reference to this Builder enabling method chaining.
+ *
+ * @param dcCount the {@code dcCount} to set
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration dcCount(int dcCount)
+ {
+ this.dcCount = dcCount;
+ return this;
+ }
+
+ /**
+ * Sets the {@code newNodesPerDc} and returns a reference to this Builder enabling method chaining.
+ *
+ * @param newNodesPerDc the {@code newNodesPerDc} to set
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration newNodesPerDc(int newNodesPerDc)
+ {
+ Preconditions.checkArgument(newNodesPerDc >= 0,
+ "newNodesPerDc cannot be a negative number");
+ this.newNodesPerDc = newNodesPerDc;
+ return this;
+ }
+
+ /**
+ * Sets the {@code numDataDirsPerInstance} and returns a reference to this Builder enabling method chaining.
+ *
+ * @param numDataDirsPerInstance the {@code numDataDirsPerInstance} to set
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration numDataDirsPerInstance(int numDataDirsPerInstance)
+ {
+ this.numDataDirsPerInstance = numDataDirsPerInstance;
+ return this;
+ }
+
+ /**
+ * Sets the {@code instanceInitializer} and returns a reference to this Builder enabling method chaining.
+ *
+ * @param instanceInitializer the {@code instanceInitializer} to set
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration instanceInitializer(BiConsumer<ClassLoader, Integer> instanceInitializer)
+ {
+ this.instanceInitializer = instanceInitializer;
+ return this;
+ }
+
+ /**
+ * Sets the {@code dynamicPortAllocation} and returns a reference to this Builder enabling method chaining.
+ *
+ * @param dynamicPortAllocation the {@code dynamicPortAllocation} to set
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration dynamicPortAllocation(boolean dynamicPortAllocation)
+ {
+ this.dynamicPortAllocation = dynamicPortAllocation;
+ return this;
+ }
+}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/IClusterExtension.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/IClusterExtension.java
new file mode 100644
index 0000000..c604e37
--- /dev/null
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/IClusterExtension.java
@@ -0,0 +1,102 @@
+/*
+ * 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 org.apache.cassandra.testing;
+
+import java.util.function.Consumer;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+
+/**
+ * Extends functionality for the {@link ICluster} interface
+ *
+ * @param <I> the type of the cluster instances
+ */
+public interface IClusterExtension<I extends IInstance> extends ICluster<I>
+{
+ /**
+ * Change the schema of the cluster, tolerating stopped nodes. N.B. the schema
+ * will not automatically be updated when stopped nodes are restarted, individual tests need to
+ * re-synchronize somehow (by gossip or some other mechanism).
+ *
+ * @param query Schema altering statement
+ */
+ void schemaChangeIgnoringStoppedInstances(String query);
+
+ /**
+ * Create a new instance and add it to the cluster, without starting it.
+ *
+ * @param dc the instance should be in
+ * @param rack the instance should be in
+ * @param fn function to add to the config before starting
+ * @return the instance added
+ */
+ I addInstance(String dc, String rack, Consumer<IInstanceConfig> fn);
+
+ /**
+ * @return the first instance with running state
+ */
+ I getFirstRunningInstance();
+
+ /**
+ * @return a newly created instance configuration
+ */
+ IInstanceConfig newInstanceConfig();
+
+ /**
+ * @param nodeNum the node number
+ * @return a newly created instances configuration for the node {@code nodeNum}
+ */
+ IInstanceConfig createInstanceConfig(int nodeNum);
+
+ /**
+ * @return a reference to the delegated {@link ICluster} instance
+ */
+ ICluster<I> delegate();
+
+ /**
+ * Waits for the ring to have the target instance with the provided state.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing to look for
+ * @param state expected
+ */
+ void awaitRingState(IInstance instance, IInstance expectedInRing, String state);
+
+ /**
+ * Waits for the target instance to have the desired status. Target status is checked via string contains so works
+ * with 'NORMAL' but also can check tokens or full state.
+ *
+ * @param instance instance to check on
+ * @param expectedInGossip instance to wait for
+ * @param targetStatus for the instance
+ */
+ void awaitGossipStatus(IInstance instance, IInstance expectedInGossip, String targetStatus);
+
+ /**
+ * Stop an instance in a blocking manner.
+ * <p>The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+ * the exceptions and throw as runtime.
+ *
+ * @param instance instance to stop
+ */
+ void stopUnchecked(IInstance instance);
+}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/IsolatedDTestClassLoaderWrapper.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/IsolatedDTestClassLoaderWrapper.java
new file mode 100644
index 0000000..728d804
--- /dev/null
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/IsolatedDTestClassLoaderWrapper.java
@@ -0,0 +1,210 @@
+/*
+ * 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 org.apache.cassandra.testing;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.lang.reflect.Constructor;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.vdurmont.semver4j.Semver;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.shared.Versions;
+import org.apache.cassandra.sidecar.testing.SharedClusterIntegrationTestBase;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Wraps functionality for the DTestJarClassLoader to be shared with the {@link SharedClusterIntegrationTestBase}
+ */
+public class IsolatedDTestClassLoaderWrapper
+{
+ protected DTestJarClassLoader dtestJarClassLoader;
+
+ public void initializeDTestJarClassLoader(TestVersion testVersion, Class<?> clazz)
+ {
+ ClassLoader parent = Thread.currentThread().getContextClassLoader();
+ Semver version = new Semver(testVersion.version(), Semver.SemverType.LOOSE);
+ List<URL> urlList = new ArrayList<>(Arrays.asList(Versions.find().getLatest(version).classpath));
+ URL classUrl = urlOfClass(clazz);
+ urlList.add(classUrl);
+ dtestJarClassLoader = new DTestJarClassLoader(urlList.toArray(new URL[0]), parent);
+ }
+
+ public void closeDTestJarClassLoader()
+ {
+ if (dtestJarClassLoader != null)
+ {
+ try
+ {
+ dtestJarClassLoader.close();
+ }
+ catch (IOException ioException)
+ {
+ throw new UncheckedIOException("Failed to close custom class loader", ioException);
+ }
+ }
+ }
+
+ public <T> T executeActionOnDTestClassLoader(ExecutableAction<T> action)
+ {
+ Thread currentThread = Thread.currentThread();
+ ClassLoader originalClassLoader = currentThread.getContextClassLoader();
+ try
+ {
+ currentThread.setContextClassLoader(dtestJarClassLoader);
+ return action.run();
+ }
+ finally
+ {
+ currentThread.setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ public <T> void executeExceptionableActionOnDTestClassLoader(ExecutableExceptionableAction<T> action) throws IOException
+ {
+ Thread currentThread = Thread.currentThread();
+ ClassLoader originalClassLoader = currentThread.getContextClassLoader();
+
+ try
+ {
+ currentThread.setContextClassLoader(dtestJarClassLoader);
+ action.run();
+ }
+ finally
+ {
+ currentThread.setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public IClusterExtension<? extends IInstance> loadCluster(String versionString,
+ ClusterBuilderConfiguration builderConfiguration)
+ {
+ return executeActionOnDTestClassLoader(() -> {
+ try
+ {
+ Class<?> launcherClass = Class.forName("org.apache.cassandra.distributed.impl.CassandraCluster", true, dtestJarClassLoader);
+ Constructor<IClusterExtension<? extends IInstance>> ctor =
+ (Constructor<IClusterExtension<? extends IInstance>>) launcherClass.getDeclaredConstructor(String.class,
+ ClusterBuilderConfiguration.class);
+ return ctor.newInstance(versionString, builderConfiguration);
+ }
+ catch (ReflectiveOperationException e)
+ {
+ throw new RuntimeException("Unable to provision cluster for version " + versionString, e);
+ }
+ });
+ }
+
+ /**
+ * Inner class loader to isolate the dtest jar classes
+ */
+ @VisibleForTesting
+ public static class DTestJarClassLoader extends URLClassLoader
+ {
+ DTestJarClassLoader(URL[] urls, ClassLoader parent)
+ {
+ super(urls, parent);
+ }
+
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException
+ {
+ if (!"org.apache.cassandra.distributed.impl.CassandraCluster".equals(name))
+ {
+ return super.loadClass(name, resolve);
+ }
+
+ // First, check if the class has already been loaded
+ Class<?> type = findLoadedClass(name);
+ if (type == null)
+ {
+ try
+ {
+ type = findClass(name);
+ }
+ catch (ClassNotFoundException | SecurityException | LinkageError exception)
+ {
+ // ClassNotFoundException thrown if class not found
+ }
+ if (type == null)
+ {
+ // If not found, then invoke findClass in order to find the class
+ type = super.loadClass(name, false);
+ }
+ }
+ if (resolve)
+ {
+ resolveClass(type);
+ }
+ return type;
+ }
+ }
+
+ static URL urlOfClass(Class<?> clazz)
+ {
+ URL classUrl = clazz.getResource(clazz.getSimpleName() + ".class");
+ assertThat(classUrl).isNotNull();
+ String pathOfClassInJar = "!/" + clazz.getCanonicalName().replace(".", "/") + ".class";
+ String classDir = classUrl.getPath().replace(pathOfClassInJar, "");
+ try
+ {
+ return new URL(classDir);
+ }
+ catch (MalformedURLException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Encapsulates action to be executed inside the classloader
+ *
+ * @param <T> return type
+ */
+ public interface ExecutableAction<T>
+ {
+ /**
+ * @return (optional) result from the operation that runs inside the classloader
+ */
+ T run();
+ }
+
+ /**
+ * Encapsulates an action that can throw an {@link IOException} while executing inside the classloader
+ *
+ * @param <T> return type
+ */
+ public interface ExecutableExceptionableAction<T>
+ {
+ /**
+ * @return (optional) result from the operation that runs inside the classloader
+ * @throws IOException when an IO exception occurs while running this operation
+ */
+ T run() throws IOException;
+ }
+}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TestTokenSupplier.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestTokenSupplier.java
similarity index 94%
rename from cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TestTokenSupplier.java
rename to cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestTokenSupplier.java
index 6d23033..28e8e4f 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TestTokenSupplier.java
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestTokenSupplier.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.cassandra.analytics;
+package org.apache.cassandra.testing;
import java.math.BigInteger;
import java.util.ArrayList;
@@ -39,8 +39,10 @@ public class TestTokenSupplier
* Tokens are allocation used in tests to simulate token allocation nodes for an approx even distribution
* in a multiDC environment with neighboring nodes different DCs allocated adjacent tokens.
* Allocations for new nodes are interleaved among existing tokens.
- * @param numNodesPerDC no. nodes from a single DC
- * @param numDcs no. of datacenters
+ *
+ * @param numNodesPerDC number of nodes from a single DC
+ * @param newNodesPerDC number of additional nodes for a DC
+ * @param numDcs no. of datacenters
* @param numTokensPerNode no. tokens allocated to each node (this is always 1 if there are no vnodes)
* @return The token supplier that vends the tokens
*/
@@ -60,7 +62,7 @@ public class TestTokenSupplier
// Initial value of the first new node
- BigInteger value = new BigInteger(tokens[0 + (numDcs - 1)].get(0));
+ BigInteger value = new BigInteger(tokens[(numDcs - 1)].get(0));
BigInteger subIncrement = increment.divide(BigInteger.valueOf(2));
int nodeId = (int) totalTokens + 1;
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java
index 444ac90..3527145 100644
--- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java
@@ -69,6 +69,35 @@ public final class TestUtils
return new QualifiedName(keyspace, testTablePrefix + TEST_TABLE_ID.getAndIncrement());
}
+ /**
+ * @param keyspace the name of the keyspace
+ * @param tablePrefix the prefix for the table
+ * @return a {@link QualifiedName} with quoted keyspace and quoted table
+ */
+ public static QualifiedName uniqueTestQuotedKeyspaceQuotedTableFullName(String keyspace, String tablePrefix)
+ {
+ return new QualifiedName(keyspace, tablePrefix + TEST_TABLE_ID.getAndIncrement(), true, true);
+ }
+
+ /**
+ * @param keyspace the name of the keyspace
+ * @param tablePrefix the prefix for the table
+ * @return a {@link QualifiedName} with unquoted keyspace and quoted table
+ */
+ public static QualifiedName uniqueTestKeyspaceQuotedTableFullName(String keyspace, String tablePrefix)
+ {
+ return new QualifiedName(keyspace, tablePrefix + TEST_TABLE_ID.getAndIncrement(), false, true);
+ }
+
+ /**
+ * @param keyspace the name of the keyspace
+ * @return a {@link QualifiedName} with quoted keyspace and unquoted table
+ */
+ public static QualifiedName uniqueTestQuotedKeyspaceTableFullName(String keyspace)
+ {
+ return new QualifiedName(keyspace, TEST_TABLE_PREFIX + TEST_TABLE_ID.getAndIncrement(), true, false);
+ }
+
/**
* Defaults to run in-jvm dtest
*/
@@ -98,5 +127,7 @@ public final class TestUtils
// this property is used to allow non-members of the ring to exist in gossip without breaking RF changes
// it would be nice not to rely on this, but hopefully we'll have consistent range movements before it matters
System.setProperty("cassandra.allow_alter_rf_during_range_movement", "true");
+
+ System.setProperty("cassandra.minimum_replication_factor", "1");
}
}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/utils/ClusterUtils.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/utils/ClusterUtils.java
new file mode 100644
index 0000000..e6b1449
--- /dev/null
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/utils/ClusterUtils.java
@@ -0,0 +1,183 @@
+/*
+ * 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 org.apache.cassandra.testing.utils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * <p>This class should never be called from within the cluster, always in the App ClassLoader.
+ *
+ * <p>Parts of this class are copied from Cassandra's {@code org.apache.cassandra.distributed.shared.ClusterUtils}
+ * class
+ */
+public final class ClusterUtils
+{
+ private ClusterUtils()
+ {
+ throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated");
+ }
+
+ /**
+ * Start the instance with the given System Properties, after the instance has started,
+ * the properties will be cleared.
+ *
+ * @param instance the instance
+ * @param fn the consumer function
+ * @param <I> the concrete instance type
+ * @return the started instance
+ */
+ public static <I extends IInstance> I start(I instance, Consumer<WithProperties> fn)
+ {
+ return start(instance, (ignore, prop) -> fn.accept(prop));
+ }
+
+ /**
+ * Start the instance with the given System Properties, after the instance has started,
+ * the properties will be cleared.
+ *
+ * @param instance the instance
+ * @param fn the bi-consumer function
+ * @param <I> the concrete instance type
+ * @return the started instance
+ */
+ public static <I extends IInstance> I start(I instance, BiConsumer<I, WithProperties> fn)
+ {
+ try (WithProperties properties = new WithProperties())
+ {
+ fn.accept(instance, properties);
+ instance.startup();
+ return instance;
+ }
+ }
+
+ /**
+ * Get the ring from the perspective of the instance.
+ *
+ * @param instance the instance
+ * @return a list with the parsed ring results
+ */
+ public static List<RingInstanceDetails> ring(IInstance instance)
+ {
+ NodeToolResult results = instance.nodetoolResult("ring");
+ results.asserts().success();
+ return parseRing(results.getStdout());
+ }
+
+
+ private static List<RingInstanceDetails> parseRing(String str)
+ {
+ // 127.0.0.3 rack0 Up Normal 46.21 KB 100.00% -1
+ // /127.0.0.1:7012 Unknown ? Normal ? 100.00% -3074457345618258603
+ Pattern pattern = Pattern.compile("^(/?[0-9.:]+)\\s+(\\w+|\\?)\\s+(\\w+|\\?)\\s+(\\w+|\\?).*?(-?\\d+)\\s*$");
+ List<RingInstanceDetails> details = new ArrayList<>();
+ String[] lines = str.split("\n");
+ for (String line : lines)
+ {
+ Matcher matcher = pattern.matcher(line);
+ if (!matcher.find())
+ {
+ continue;
+ }
+ details.add(new RingInstanceDetails(matcher.group(1), matcher.group(2), matcher.group(3), matcher.group(4), matcher.group(5)));
+ }
+
+ return details;
+ }
+
+
+ public static final class RingInstanceDetails
+ {
+ private final String address;
+ private final String rack;
+ private final String status;
+ private final String state;
+ private final String token;
+
+ private RingInstanceDetails(String address, String rack, String status, String state, String token)
+ {
+ this.address = address;
+ this.rack = rack;
+ this.status = status;
+ this.state = state;
+ this.token = token;
+ }
+
+ public String getAddress()
+ {
+ return address;
+ }
+
+ public String getStatus()
+ {
+ return status;
+ }
+
+ public String getState()
+ {
+ return state;
+ }
+
+ public String getToken()
+ {
+ return token;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+ RingInstanceDetails that = (RingInstanceDetails) o;
+ return Objects.equals(address, that.address) &&
+ Objects.equals(rack, that.rack) &&
+ Objects.equals(status, that.status) &&
+ Objects.equals(state, that.state) &&
+ Objects.equals(token, that.token);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(address, rack, status, state, token);
+ }
+
+ public String toString()
+ {
+ return Arrays.asList(address, rack, status, state, token).toString();
+ }
+ }
+}
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/utils/WithProperties.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/utils/WithProperties.java
new file mode 100644
index 0000000..30362eb
--- /dev/null
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/utils/WithProperties.java
@@ -0,0 +1,56 @@
+/*
+ * 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 org.apache.cassandra.testing.utils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Parts of this class are copied from Cassandra's {@code org.apache.cassandra.distributed.shared.WithProperties}
+ * class.
+ */
+public class WithProperties implements AutoCloseable
+{
+ private final List<Runnable> rollback = new ArrayList<>();
+
+ public WithProperties with(String... kvs)
+ {
+ assert kvs.length % 2 == 0 : "Input must have an even amount of inputs but given " + kvs.length;
+ for (int i = 0; i <= kvs.length - 2; i = i + 2)
+ {
+ with(kvs[i], kvs[i + 1]);
+ }
+ return this;
+ }
+
+ private void with(String key, String value)
+ {
+ String previous = System.setProperty(key, value);
+ rollback.add(previous == null ? () -> System.clearProperty(key) : () -> System.setProperty(key, previous));
+ }
+
+ @Override
+ public void close()
+ {
+ Collections.reverse(rollback);
+ rollback.forEach(Runnable::run);
+ rollback.clear();
+ }
+}
diff --git a/cassandra-analytics-integration-tests/build.gradle b/cassandra-analytics-integration-tests/build.gradle
index 38cad7e..3230831 100644
--- a/cassandra-analytics-integration-tests/build.gradle
+++ b/cassandra-analytics-integration-tests/build.gradle
@@ -19,93 +19,117 @@ import java.nio.file.Paths
* under the License.
*/
-project(':cassandra-analytics-integration-tests') {
- apply(plugin: 'java-library')
-
- def integrationMaxHeapSize = System.getenv("INTEGRATION_MAX_HEAP_SIZE") ?: "3000M"
- println("Using ${integrationMaxHeapSize} maxHeapSize")
-
- def integrationMaxParallelForks = (System.getenv("INTEGRATION_MAX_PARALLEL_FORKS") ?: "4") as int
- println("Using ${integrationMaxParallelForks} maxParallelForks")
+plugins {
+ id('idea')
+ // A Java library
+ id('java-library')
+ // which is published
+ id('maven-publish')
+}
- configurations {
- all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
- all*.exclude(group: 'log4j', module: 'log4j')
+publishing {
+ publications {
+ maven(MavenPublication) {
+ from components.java
+ groupId project.group
+ artifactId "${archivesBaseName}"
+ version System.getenv("CODE_VERSION") ?: "${version}"
+ }
}
- dependencies {
-
- testImplementation(project(':cassandra-bridge'))
- testImplementation(project(':cassandra-analytics-core'))
- testImplementation(group: 'net.java.dev.jna', name: 'jna', version: '5.9.0')
+}
- testImplementation(group: "${sparkGroupId}", name: "spark-core_${scalaMajorVersion}", version: "${sparkVersion}")
- testImplementation(group: "${sparkGroupId}", name: "spark-sql_${scalaMajorVersion}", version: "${sparkVersion}")
+def integrationMaxHeapSize = System.getenv("INTEGRATION_MAX_HEAP_SIZE") ?: "3000M"
+println("Using ${integrationMaxHeapSize} maxHeapSize")
+
+def integrationMaxParallelForks = (System.getenv("INTEGRATION_MAX_PARALLEL_FORKS") ?: "4") as int
+println("Using ${integrationMaxParallelForks} maxParallelForks")
+
+configurations {
+ // remove netty-all dependency coming from spark
+ all*.exclude(group: 'io.netty', module: 'netty-all')
+ configureEach {
+ resolutionStrategy {
+ // Force set the vertx versions that do not depend on a newer jackson version
+ force("io.vertx:vertx-web:${vertxVersion}")
+ force("io.vertx:vertx-core:${vertxVersion}")
+ force("io.vertx:vertx-auth-common:${vertxVersion}")
+ force("io.vertx:vertx-dropwizard-metrics:${vertxVersion}")
+ force("io.vertx:vertx-web-client:${vertxVersion}")
+ }
+ }
+}
- testImplementation(group: "${sidecarClientGroup}", name: "${sidecarClientName}", version: "${sidecarVersion}")
+dependencies {
+ testImplementation(project(':cassandra-analytics-core'))
+ testImplementation(group: 'net.java.dev.jna', name: 'jna', version: '5.9.0')
- // JUnit
- testImplementation("org.junit.jupiter:junit-jupiter-api:${project.junitVersion}")
- testImplementation("org.junit.jupiter:junit-jupiter-params:${project.junitVersion}")
- testImplementation("org.junit.jupiter:junit-jupiter-engine:${project.junitVersion}")
- testImplementation("org.assertj:assertj-core:3.24.2")
- testImplementation(project(path: ":cassandra-analytics-integration-framework", configuration: "shadow"))
- ext.dtestJar = System.getenv("DTEST_JAR") ?: "dtest-4.1.4.jar" // latest supported Cassandra build is 4.1
- def dtestJarFullPath = "${dependencyLocation}${ext.dtestJar}"
- println("Using DTest jar: " + dtestJarFullPath)
- testCompileOnly(files("${dtestJarFullPath}"))
- testImplementation("org.apache.cassandra:dtest-api:0.0.16")
- // Needed by the Cassandra dtest framework
- testImplementation("org.junit.vintage:junit-vintage-engine:${junitVersion}")
- testImplementation('com.datastax.cassandra:cassandra-driver-core:3.9.0')
+ testImplementation(group: "${sparkGroupId}", name: "spark-core_${scalaMajorVersion}", version: "${sparkVersion}")
+ testImplementation(group: "${sparkGroupId}", name: "spark-sql_${scalaMajorVersion}", version: "${sparkVersion}")
- testImplementation('org.mockito:mockito-inline:4.10.0')
- testImplementation("org.assertj:assertj-core:3.24.2")
+ testImplementation(project(path: ":cassandra-analytics-integration-framework"))
+}
- // Logging
- testImplementation(group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.22.0')
- testImplementation(group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.22.0')
- testImplementation(group: 'org.apache.logging.log4j', name: 'log4j-slf4j-impl', version: '2.22.0')
- testImplementation(group: 'org.apache.logging.log4j', name: 'log4j-1.2-api', version: '2.22.0')
+test {
+ // Because system properties aren't passed from the command line through to tests, we need to specifically
+ // set them again here.
+ systemProperty "cassandra.test.dtest_jar_path", dependencyLocation
+ systemProperty "cassandra.sidecar.versions_to_test",
+ System.getProperty("cassandra.sidecar.versions_to_test", "4.1")
+ systemProperty "SKIP_STARTUP_VALIDATIONS", "true"
+ systemProperty "logback.configurationFile", "src/test/resources/logback-test.xml"
+ minHeapSize = '1g'
+ maxHeapSize = integrationMaxHeapSize
+ maxParallelForks = integrationMaxParallelForks
+ forkEvery = 1 // Enables different end-to-end test classes use Spark contexts with different configurations
+
+ beforeTest { descriptor ->
+ // println("Setting test tags to ${descriptor.className}-${descriptor.name}")
+ systemProperty "cassandra.testtag", "${descriptor.className}"
+ systemProperty "suitename", "${descriptor.name}"
}
+ testLogging {
+ events "passed", "skipped", "failed"
+ }
- test {
- // Because system properties aren't passed from the command line through to tests, we need to specifically
- // set them again here.
- systemProperty "cassandra.test.dtest_jar_path", dependencyLocation
- systemProperty "cassandra.sidecar.versions_to_test",
- System.getProperty("cassandra.sidecar.versions_to_test", "4.1")
- systemProperty "SKIP_STARTUP_VALIDATIONS", "true"
- systemProperty "logback.configurationFile", "src/test/resources/logback-test.xml"
- minHeapSize = '1g'
- maxHeapSize = integrationMaxHeapSize
- maxParallelForks = integrationMaxParallelForks
- forkEvery = 1 // Enables different end-to-end test classes use Spark contexts with different configurations
+ if (JavaVersion.current().isJava11Compatible()) {
+ def JDK11_OPTIONS = ['-Djdk.attach.allowAttachSelf=true',
+ '--add-exports', 'java.base/jdk.internal.misc=ALL-UNNAMED',
+ '--add-exports', 'java.base/jdk.internal.ref=ALL-UNNAMED',
+ '--add-exports', 'java.base/sun.nio.ch=ALL-UNNAMED',
+ '--add-exports', 'java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED',
+ '--add-exports', 'java.rmi/sun.rmi.registry=ALL-UNNAMED',
+ '--add-exports', 'java.rmi/sun.rmi.server=ALL-UNNAMED',
+ '--add-exports', 'java.sql/java.sql=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.lang.module=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.loader=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.ref=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.reflect=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.math=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.module=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.util.jar=ALL-UNNAMED',
+ '--add-opens', 'jdk.management/com.sun.management.internal=ALL-UNNAMED']
+ jvmArgs(JDK11_OPTIONS)
+ println("JVM arguments for $project.name are $allJvmArgs")
+ }
- beforeTest {descriptor ->
- // println("Setting test tags to ${descriptor.className}-${descriptor.name}")
- systemProperty "cassandra.testtag", "${descriptor.className}"
- systemProperty "suitename", "${descriptor.name}"
+ useJUnitPlatform()
+ def destDir = Paths.get(rootProject.rootDir.absolutePath, "build", "test-reports", "integration").toFile()
+ reports {
+ junitXml {
+ enabled true
+ destination = destDir
}
-
- testLogging {
- events "passed", "skipped", "failed"
+ html {
+ enabled true
+ destination = destDir
}
+ }
+}
- // Make it so unit tests run on a jar with Cassandra bridge implementations built in
- dependsOn(tasks.jar)
- classpath = project.sourceSets.test.output + configurations.testRuntimeClasspath + files(jar.archiveFile)
- useJUnitPlatform()
- def destDir = Paths.get(rootProject.rootDir.absolutePath, "build", "test-reports", "integration").toFile()
- reports {
- junitXml {
- enabled true
- destination = destDir
- }
- html {
- enabled true
- destination = destDir
- }
- }
+java {
+ // Allows publishing the test jar
+ registerFeature("test") {
+ usingSourceSet(sourceSets.test)
}
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BlockedInstancesTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BlockedInstancesTest.java
index 02f3320..83e1ad7 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BlockedInstancesTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BlockedInstancesTest.java
@@ -18,25 +18,25 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
+import org.jetbrains.annotations.NotNull;
import static com.datastax.driver.core.ConsistencyLevel.ALL;
import static com.datastax.driver.core.ConsistencyLevel.ONE;
@@ -51,7 +51,7 @@ import static org.assertj.core.api.Assertions.catchThrowable;
/**
* Collection of in-jvm-dtest based integration tests to validate the bulk write path when blocked instances
* are configured.
- *
+ * <p>
* Note: Since these tests are derived from {@link SharedClusterSparkIntegrationTestBase}, the tests reuse the same
* cluster with a unique table per test case. The implicit mapping from the test case to the table is made by
* creating the tables using the method name in {@link BlockedInstancesTest#initializeSchemaForTest()}.
@@ -65,7 +65,7 @@ public class BlockedInstancesTest extends ResiliencyTestBase
void testSingleBlockedNodeSucceeds(TestInfo testInfo)
{
TestConsistencyLevel cl = TestConsistencyLevel.of(QUORUM, QUORUM);
- QualifiedName table = new QualifiedName(TEST_KEYSPACE, testInfo.getTestMethod().get().getName().toLowerCase());
+ QualifiedName table = new QualifiedName(TEST_KEYSPACE, tableName(testInfo));
bulkWriterDataFrameWriter(df, table).option(WriterOptions.BULK_WRITER_CL.name(), cl.writeCL.name())
.option(WriterOptions.BLOCKED_CASSANDRA_INSTANCES.name(), "127.0.0.2")
.save();
@@ -81,7 +81,7 @@ public class BlockedInstancesTest extends ResiliencyTestBase
void testSingleBlockedNodeWithWriteFailure(TestInfo testInfo)
{
TestConsistencyLevel cl = TestConsistencyLevel.of(ONE, ALL);
- QualifiedName table = new QualifiedName(TEST_KEYSPACE, testInfo.getTestMethod().get().getName().toLowerCase());
+ QualifiedName table = new QualifiedName(TEST_KEYSPACE, tableName(testInfo));
Throwable thrown = catchThrowable(() ->
bulkWriterDataFrameWriter(df, table).option(WriterOptions.BULK_WRITER_CL.name(), cl.writeCL.name())
.option(WriterOptions.BLOCKED_CASSANDRA_INSTANCES.name(), "127.0.0.2")
@@ -93,7 +93,7 @@ public class BlockedInstancesTest extends ResiliencyTestBase
void testMultipleBlockedNodesWithWriteFailure(TestInfo testInfo)
{
TestConsistencyLevel cl = TestConsistencyLevel.of(QUORUM, QUORUM);
- QualifiedName table = new QualifiedName(TEST_KEYSPACE, testInfo.getTestMethod().get().getName().toLowerCase());
+ QualifiedName table = new QualifiedName(TEST_KEYSPACE, tableName(testInfo));
Throwable thrown = catchThrowable(() ->
bulkWriterDataFrameWriter(df, table).option(WriterOptions.BULK_WRITER_CL.name(), cl.writeCL.name())
.option(WriterOptions.BLOCKED_CASSANDRA_INSTANCES.name(), "127.0.0.2,127.0.0.3")
@@ -114,17 +114,15 @@ public class BlockedInstancesTest extends ResiliencyTestBase
assertThat(cause).isNotNull()
.hasMessageFindingMatch(String.format("Failed to load (\\d+) ranges with %s for " +
- "job ([a-zA-Z0-9-]+) in phase Environment Validation.", cl.writeCL));
+ "job ([a-zA-Z0-9-]+) in phase Environment Validation.", cl.writeCL));
expectedInstanceData.entrySet()
- .stream()
.forEach(e -> e.setValue(Collections.emptySet()));
validateNodeSpecificData(table, expectedInstanceData);
validateData(table, cl.readCL, 0);
-
}
-
+ @Override
protected ClusterBuilderConfiguration testClusterConfiguration()
{
return clusterConfig().nodesPerDc(3)
@@ -134,6 +132,7 @@ public class BlockedInstancesTest extends ResiliencyTestBase
@Override
protected void beforeTestStart()
{
+ super.beforeTestStart();
SparkSession spark = getOrCreateSparkSession();
// Generate some artificial data for the test
df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT);
@@ -141,12 +140,6 @@ public class BlockedInstancesTest extends ResiliencyTestBase
expectedInstanceData = generateExpectedInstanceData(cluster, Collections.emptyList(), ROW_COUNT);
}
- @Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
- {
- return clusterBuilder(testClusterConfiguration(), testVersion);
- }
-
@Override
protected void initializeSchemaForTest()
{
@@ -160,4 +153,12 @@ public class BlockedInstancesTest extends ResiliencyTestBase
}
}
}
+
+ @NotNull
+ private String tableName(TestInfo testInfo)
+ {
+ Optional<Method> testMethod = testInfo.getTestMethod();
+ assertThat(testMethod).isPresent();
+ return testMethod.get().getName().toLowerCase();
+ }
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteTtlTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteTtlTest.java
index 6d58e7d..d5d198e 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteTtlTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteTtlTest.java
@@ -19,30 +19,21 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
-import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.Uninterruptibles;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.TTLOption;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
-import static org.apache.cassandra.analytics.ResiliencyTestBase.fixDistributedSchemas;
-import static org.apache.cassandra.analytics.ResiliencyTestBase.waitForHealthyRing;
import static org.apache.cassandra.testing.TestUtils.DC1_RF3;
import static org.apache.cassandra.testing.TestUtils.ROW_COUNT;
import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
@@ -97,30 +88,10 @@ class BulkWriteTtlTest extends SharedClusterSparkIntegrationTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected ClusterBuilderConfiguration testClusterConfiguration()
{
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(3)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(3, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- UpgradeableCluster cluster = clusterBuilder.createWithoutStarting();
- cluster.startup();
-
- waitForHealthyRing(cluster);
- fixDistributedSchemas(cluster);
- return cluster;
+ return super.testClusterConfiguration()
+ .nodesPerDc(3);
}
@Override
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteUdtTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteUdtTest.java
index 481e4ef..47ad6cf 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteUdtTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteUdtTest.java
@@ -19,27 +19,17 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
-
import org.junit.jupiter.api.Test;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.jetbrains.annotations.NotNull;
-import static org.apache.cassandra.analytics.ResiliencyTestBase.fixDistributedSchemas;
-import static org.apache.cassandra.analytics.ResiliencyTestBase.waitForHealthyRing;
import static org.apache.cassandra.testing.TestUtils.DC1_RF3;
import static org.apache.cassandra.testing.TestUtils.ROW_COUNT;
import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
@@ -78,8 +68,7 @@ class BulkWriteUdtTest extends SharedClusterSparkIntegrationTestBase
SimpleQueryResult result = cluster.coordinator(1).executeWithResult("SELECT * FROM " + UDT_TABLE_NAME, ConsistencyLevel.ALL);
assertThat(result.hasNext()).isTrue();
validateWritesWithDriverResultSet(df.collectAsList(),
- queryAllDataWithDriver(cluster, UDT_TABLE_NAME,
- relocated.shaded.com.datastax.driver.core.ConsistencyLevel.LOCAL_QUORUM),
+ queryAllDataWithDriver(UDT_TABLE_NAME),
BulkWriteUdtTest::defaultRowFormatter);
}
@@ -94,13 +83,12 @@ class BulkWriteUdtTest extends SharedClusterSparkIntegrationTestBase
SimpleQueryResult result = cluster.coordinator(1).executeWithResult("SELECT * FROM " + NESTED_TABLE_NAME, ConsistencyLevel.ALL);
assertThat(result.hasNext()).isTrue();
validateWritesWithDriverResultSet(df.collectAsList(),
- queryAllDataWithDriver(cluster, NESTED_TABLE_NAME,
- relocated.shaded.com.datastax.driver.core.ConsistencyLevel.LOCAL_QUORUM),
+ queryAllDataWithDriver(NESTED_TABLE_NAME),
BulkWriteUdtTest::defaultRowFormatter);
}
@NotNull
- public static String defaultRowFormatter(relocated.shaded.com.datastax.driver.core.Row row)
+ public static String defaultRowFormatter(com.datastax.driver.core.Row row)
{
return row.getLong(0) +
":" +
@@ -108,30 +96,10 @@ class BulkWriteUdtTest extends SharedClusterSparkIntegrationTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected ClusterBuilderConfiguration testClusterConfiguration()
{
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(3)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(3, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- UpgradeableCluster cluster = clusterBuilder.createWithoutStarting();
- cluster.startup();
-
- waitForHealthyRing(cluster);
- fixDistributedSchemas(cluster);
- return cluster;
+ return super.testClusterConfiguration()
+ .nodesPerDc(3);
}
@Override
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersReadTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersReadTest.java
index 89513e3..7627d1e 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersReadTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersReadTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
@@ -34,21 +33,15 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
-import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.testing.TestVersion;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.cassandra.testing.TestUtils.DC1_RF1;
import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
-import static org.apache.cassandra.testing.TestUtils.uniqueTestTableFullName;
+import static org.apache.cassandra.testing.TestUtils.uniqueTestKeyspaceQuotedTableFullName;
+import static org.apache.cassandra.testing.TestUtils.uniqueTestQuotedKeyspaceQuotedTableFullName;
+import static org.apache.cassandra.testing.TestUtils.uniqueTestQuotedKeyspaceTableFullName;
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -60,12 +53,12 @@ import static org.assertj.core.api.Assertions.assertThat;
class QuoteIdentifiersReadTest extends SharedClusterSparkIntegrationTestBase
{
static final List<String> DATASET = Arrays.asList("a", "b", "c", "d", "e", "f", "g");
- static final QualifiedName TABLE_NAME_FOR_UDT_TEST = uniqueTestTableFullName("QuOtEd_KeYsPaCe", "QuOtEd_TaBlE");
+ static final QualifiedName TABLE_NAME_FOR_UDT_TEST = uniqueTestQuotedKeyspaceQuotedTableFullName("QuOtEd_KeYsPaCe", "QuOtEd_TaBlE");
static final List<QualifiedName> TABLE_NAMES =
- Arrays.asList(uniqueTestTableFullName("QuOtEd_KeYsPaCe"),
- uniqueTestTableFullName("keyspace"), // keyspace is a reserved word
- uniqueTestTableFullName(TEST_KEYSPACE, "QuOtEd_TaBlE"),
- new QualifiedName(TEST_KEYSPACE, "table"), // table is a reserved word
+ Arrays.asList(uniqueTestQuotedKeyspaceTableFullName("QuOtEd_KeYsPaCe"),
+ uniqueTestQuotedKeyspaceTableFullName("keyspace"), // keyspace is a reserved word
+ uniqueTestKeyspaceQuotedTableFullName(TEST_KEYSPACE, "QuOtEd_TaBlE"),
+ new QualifiedName(TEST_KEYSPACE, "table", false, true), // table is a reserved word
TABLE_NAME_FOR_UDT_TEST);
@ParameterizedTest(name = "{index} => table={0}")
@@ -112,30 +105,6 @@ class QuoteIdentifiersReadTest extends SharedClusterSparkIntegrationTestBase
return TABLE_NAMES.stream().map(Arguments::of);
}
- @Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
- {
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(1)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(1, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- UpgradeableCluster cluster = clusterBuilder.createWithoutStarting();
- cluster.startup();
- return cluster;
- }
-
@Override
protected void initializeSchemaForTest()
{
@@ -155,7 +124,7 @@ class QuoteIdentifiersReadTest extends SharedClusterSparkIntegrationTestBase
// Create UDT
String createUdtQuery = "CREATE TYPE " + TABLE_NAME_FOR_UDT_TEST.maybeQuotedKeyspace()
+ ".\"UdT1\" (\"TimE\" bigint, \"limit\" int);";
- cluster.schemaChangeIgnoringStoppedInstances(createUdtQuery);
+ cluster.schemaChange(createUdtQuery);
createTestTable(TABLE_NAME_FOR_UDT_TEST, "CREATE TABLE IF NOT EXISTS %s (" +
"\"IdEnTiFiEr\" text, " +
@@ -178,9 +147,7 @@ class QuoteIdentifiersReadTest extends SharedClusterSparkIntegrationTestBase
String query = String.format("INSERT INTO %s (\"IdEnTiFiEr\", IdEnTiFiEr) " +
"VALUES ('%s', %d);", tableName, value + "_" + uniqueNumberForTest,
((uniqueNumberForTest * 100) + i));
- cluster.getFirstRunningInstance()
- .coordinator()
- .execute(query, ConsistencyLevel.ALL);
+ cluster.schemaChange(query);
}
}
@@ -195,9 +162,7 @@ class QuoteIdentifiersReadTest extends SharedClusterSparkIntegrationTestBase
"VALUES ('%s', %d, { \"TimE\" : %d, \"limit\" : %d });",
tableName, value + "_" + uniqueNumberForTest,
((uniqueNumberForTest * 100) + i), i, i);
- cluster.getFirstRunningInstance()
- .coordinator()
- .execute(query, ConsistencyLevel.ALL);
+ cluster.schemaChange(query);
}
}
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersWriteTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersWriteTest.java
index 92110f3..0961c65 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersWriteTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/QuoteIdentifiersWriteTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
@@ -28,26 +27,20 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
-import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.jetbrains.annotations.NotNull;
-import relocated.shaded.com.datastax.driver.core.ConsistencyLevel;
import static org.apache.cassandra.analytics.DataGenerationUtils.generateCourseData;
import static org.apache.cassandra.testing.TestUtils.DC1_RF1;
import static org.apache.cassandra.testing.TestUtils.ROW_COUNT;
import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
-import static org.apache.cassandra.testing.TestUtils.uniqueTestTableFullName;
+import static org.apache.cassandra.testing.TestUtils.uniqueTestKeyspaceQuotedTableFullName;
+import static org.apache.cassandra.testing.TestUtils.uniqueTestQuotedKeyspaceQuotedTableFullName;
+import static org.apache.cassandra.testing.TestUtils.uniqueTestQuotedKeyspaceTableFullName;
/**
* Tests the bulk writer behavior when requiring quoted identifiers for keyspace, table name, and column names.
@@ -57,12 +50,12 @@ import static org.apache.cassandra.testing.TestUtils.uniqueTestTableFullName;
*/
class QuoteIdentifiersWriteTest extends SharedClusterSparkIntegrationTestBase
{
- static final QualifiedName TABLE_NAME_FOR_UDT_TEST = uniqueTestTableFullName("QuOtEd_KeYsPaCe", "QuOtEd_TaBlE");
+ static final QualifiedName TABLE_NAME_FOR_UDT_TEST = uniqueTestQuotedKeyspaceQuotedTableFullName("QuOtEd_KeYsPaCe", "QuOtEd_TaBlE");
static final List<QualifiedName> TABLE_NAMES =
- Arrays.asList(uniqueTestTableFullName("QuOtEd_KeYsPaCe"),
- uniqueTestTableFullName("keyspace"), // keyspace is a reserved word
- uniqueTestTableFullName(TEST_KEYSPACE, "QuOtEd_TaBlE"),
- new QualifiedName(TEST_KEYSPACE, "table"), // table is a reserved word
+ Arrays.asList(uniqueTestQuotedKeyspaceTableFullName("QuOtEd_KeYsPaCe"),
+ uniqueTestQuotedKeyspaceTableFullName("keyspace"), // keyspace is a reserved word
+ uniqueTestKeyspaceQuotedTableFullName(TEST_KEYSPACE, "QuOtEd_TaBlE"),
+ new QualifiedName(TEST_KEYSPACE, "table", false, true), // table is a reserved word
TABLE_NAME_FOR_UDT_TEST);
@ParameterizedTest(name = "{index} => table={0}")
@@ -91,24 +84,23 @@ class QuoteIdentifiersWriteTest extends SharedClusterSparkIntegrationTestBase
bulkWriterDataFrameWriter(df, tableName).option(WriterOptions.QUOTE_IDENTIFIERS.name(), "true")
.save();
validateWritesWithDriverResultSet(df.collectAsList(),
- queryAllDataWithDriver(cluster, tableName,
- ConsistencyLevel.LOCAL_QUORUM),
- udfData ?
- QuoteIdentifiersWriteTest::rowWithUdtFormatter :
- QuoteIdentifiersWriteTest::defaultRowFormatter);
+ queryAllDataWithDriver(tableName),
+ udfData ?
+ QuoteIdentifiersWriteTest::rowWithUdtFormatter :
+ QuoteIdentifiersWriteTest::defaultRowFormatter);
}
- public static String defaultRowFormatter(relocated.shaded.com.datastax.driver.core.Row row)
+ public static String defaultRowFormatter(com.datastax.driver.core.Row row)
{
return row.getInt("IdEnTiFiEr") +
- ":'" +
- row.getString("course") +
- "':" +
- row.getInt("limit");
+ ":'" +
+ row.getString("course") +
+ "':" +
+ row.getInt("limit");
}
@NotNull
- private static String rowWithUdtFormatter(relocated.shaded.com.datastax.driver.core.Row row)
+ private static String rowWithUdtFormatter(com.datastax.driver.core.Row row)
{
return row.getInt("IdEnTiFiEr") +
":'" +
@@ -124,30 +116,6 @@ class QuoteIdentifiersWriteTest extends SharedClusterSparkIntegrationTestBase
return TABLE_NAMES.stream().map(Arguments::of);
}
- @Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
- {
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(1)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(1, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- UpgradeableCluster cluster = clusterBuilder.createWithoutStarting();
- cluster.startup();
- return cluster;
- }
-
@Override
protected void initializeSchemaForTest()
{
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ReadDifferentTablesTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ReadDifferentTablesTest.java
index f046dc2..88e215d 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ReadDifferentTablesTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ReadDifferentTablesTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
@@ -27,16 +26,9 @@ import java.util.stream.Collectors;
import org.junit.jupiter.api.Test;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstance;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.testing.TestVersion;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
@@ -98,26 +90,4 @@ class ReadDifferentTablesTest extends SharedClusterSparkIntegrationTestBase
firstRunningInstance.coordinator().execute(query2, ConsistencyLevel.ALL);
}
}
-
- @Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
- {
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(1)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(1, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- return clusterBuilder.start();
- }
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ResiliencyTestBase.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ResiliencyTestBase.java
index 7024394..084fac5 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ResiliencyTestBase.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/ResiliencyTestBase.java
@@ -24,46 +24,32 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
-import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import com.google.common.base.Preconditions;
import com.google.common.collect.Range;
import com.datastax.driver.core.ConsistencyLevel;
-import com.vdurmont.semver4j.Semver;
-import o.a.c.analytics.sidecar.shaded.testing.adapters.base.StorageJmxOperations;
-import o.a.c.analytics.sidecar.shaded.testing.common.JmxClient;
-import org.apache.cassandra.distributed.UpgradeableCluster;
-import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.distributed.api.IInstanceConfig;
-import org.apache.cassandra.distributed.api.IUpgradeableInstance;
import org.apache.cassandra.distributed.api.Row;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.impl.AbstractCluster;
-import org.apache.cassandra.distributed.shared.ClusterUtils;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
+import org.apache.cassandra.sidecar.common.JmxClient;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.DecoratedKey;
import org.apache.cassandra.spark.bulkwriter.Tokenizer;
import org.apache.cassandra.spark.common.schema.ColumnType;
import org.apache.cassandra.spark.common.schema.ColumnTypes;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import scala.Tuple2;
-import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack;
-import static org.apache.cassandra.distributed.shared.NetworkTopology.networkTopology;
import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
import static org.assertj.core.api.Assertions.assertThat;
@@ -74,55 +60,6 @@ public abstract class ResiliencyTestBase extends SharedClusterSparkIntegrationTe
{
public static final String QUERY_ALL_ROWS = "SELECT * FROM %s";
- public UpgradeableCluster clusterBuilder(ClusterBuilderConfiguration configuration, TestVersion testVersion)
- throws IOException
- {
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- int nodesPerDc = configuration.nodesPerDc;
- int dcCount = configuration.dcCount;
- int newNodesPerDc = configuration.newNodesPerDc;
- Preconditions.checkArgument(newNodesPerDc >= 0,
- "newNodesPerDc cannot be a negative number");
- int originalNodeCount = nodesPerDc * dcCount;
- int finalNodeCount = dcCount * (nodesPerDc + newNodesPerDc);
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(),
- Semver.SemverType.LOOSE));
-
- TokenSupplier tokenSupplier = TestTokenSupplier.evenlyDistributedTokens(nodesPerDc, newNodesPerDc, dcCount, 1);
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(originalNodeCount)
- .withDynamicPortAllocation(true) // to allow parallel test runs
- .withVersion(requestedVersion)
- .withDCs(dcCount)
- .withDataDirCount(configuration.numDataDirsPerInstance)
- .withConfig(config -> configuration.features.forEach(config::with))
- .withTokenSupplier(tokenSupplier)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE);
-
- if (dcCount > 1)
- {
- clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount,
- (nodeId) -> nodeId % 2 != 0 ?
- dcAndRack("datacenter1", "rack1") :
- dcAndRack("datacenter2", "rack2")));
- }
-
- if (configuration.instanceInitializer != null)
- {
- clusterBuilder.withInstanceInitializer(configuration.instanceInitializer);
- }
-
- UpgradeableCluster cluster = clusterBuilder.start();
- if (cluster.size() > 1)
- {
- waitForHealthyRing(cluster);
- fixDistributedSchemas(cluster);
- }
- return cluster;
- }
-
public Set<String> getDataForRange(Range<BigInteger> range, int rowCount)
{
// Iterate through all data entries; filter only entries that belong to range; convert to strings
@@ -168,7 +105,7 @@ public abstract class ResiliencyTestBase extends SharedClusterSparkIntegrationTe
/**
* Returns the expected set of rows as strings for each instance in the cluster
*/
- public Map<IInstance, Set<String>> generateExpectedInstanceData(AbstractCluster<? extends IInstance> cluster,
+ public Map<IInstance, Set<String>> generateExpectedInstanceData(ICluster<? extends IInstance> cluster,
List<? extends IInstance> pendingNodes,
int rowCount)
{
@@ -252,13 +189,11 @@ public abstract class ResiliencyTestBase extends SharedClusterSparkIntegrationTe
{
IInstanceConfig config = instance.config();
Map<List<String>, List<String>> ranges = null;
- try (JmxClient client = JmxClient.builder()
- .host(config.broadcastAddress().getAddress().getHostAddress())
- .port(config.jmxPort())
- .build())
+ try (JmxClient client = wrapJmxClient(JmxClient.builder()
+ .host(config.broadcastAddress().getAddress().getHostAddress())
+ .port(config.jmxPort())))
{
- StorageJmxOperations ss = client.proxy(StorageJmxOperations.class,
- "org.apache.cassandra.db:type=StorageService");
+ SSProxy ss = client.proxy(SSProxy.class, "org.apache.cassandra.db:type=StorageService");
ranges = isPending ? ss.getPendingRangeToEndpointWithPortMap(TEST_KEYSPACE)
: ss.getRangeToEndpointWithPortMap(TEST_KEYSPACE);
@@ -306,109 +241,6 @@ public abstract class ResiliencyTestBase extends SharedClusterSparkIntegrationTe
return new ClusterBuilderConfiguration();
}
- /**
- * {@code ClusterBuilder} static inner class.
- */
- public static class ClusterBuilderConfiguration
- {
- public int nodesPerDc = 1;
- public int dcCount = 1;
- public int newNodesPerDc = 0;
- public int numDataDirsPerInstance = 1;
- private final EnumSet<Feature> features = EnumSet.of(Feature.GOSSIP, Feature.JMX, Feature.NATIVE_PROTOCOL);
- public BiConsumer<ClassLoader, Integer> instanceInitializer = null;
-
- private ClusterBuilderConfiguration()
- {
- }
-
- /**
- * Adds a features to the list of default features.
- *
- * @param feature the {@code feature} to add
- * @return a reference to this Builder
- */
- public ClusterBuilderConfiguration requestFeature(Feature feature)
- {
- features.add(feature);
- return this;
- }
-
- /**
- * Removes a feature to the list of requested features for the cluster.
- *
- * @param feature the {@code feature} to add
- * @return a reference to this Builder
- */
- public ClusterBuilderConfiguration removeFeature(Feature feature)
- {
- features.remove(feature);
- return this;
- }
-
- /**
- * Sets the {@code nodesPerDc} and returns a reference to this Builder enabling method chaining.
- *
- * @param nodesPerDc the {@code nodesPerDc} to set
- * @return a reference to this Builder
- */
- public ClusterBuilderConfiguration nodesPerDc(int nodesPerDc)
- {
- this.nodesPerDc = nodesPerDc;
- return this;
- }
-
- /**
- * Sets the {@code dcCount} and returns a reference to this Builder enabling method chaining.
- *
- * @param dcCount the {@code dcCount} to set
- * @return a reference to this Builder
- */
- public ClusterBuilderConfiguration dcCount(int dcCount)
- {
- this.dcCount = dcCount;
- return this;
- }
-
- /**
- * Sets the {@code newNodesPerDc} and returns a reference to this Builder enabling method chaining.
- *
- * @param newNodesPerDc the {@code newNodesPerDc} to set
- * @return a reference to this Builder
- */
- public ClusterBuilderConfiguration newNodesPerDc(int newNodesPerDc)
- {
- Preconditions.checkArgument(newNodesPerDc >= 0,
- "newNodesPerDc cannot be a negative number");
- this.newNodesPerDc = newNodesPerDc;
- return this;
- }
-
- /**
- * Sets the {@code numDataDirsPerInstance} and returns a reference to this Builder enabling method chaining.
- *
- * @param numDataDirsPerInstance the {@code numDataDirsPerInstance} to set
- * @return a reference to this Builder
- */
- public ClusterBuilderConfiguration numDataDirsPerInstance(int numDataDirsPerInstance)
- {
- this.numDataDirsPerInstance = numDataDirsPerInstance;
- return this;
- }
-
- /**
- * Sets the {@code instanceInitializer} and returns a reference to this Builder enabling method chaining.
- *
- * @param instanceInitializer the {@code instanceInitializer} to set
- * @return a reference to this Builder
- */
- public ClusterBuilderConfiguration instanceInitializer(BiConsumer<ClassLoader, Integer> instanceInitializer)
- {
- this.instanceInitializer = instanceInitializer;
- return this;
- }
- }
-
public static class TestConsistencyLevel
{
public final ConsistencyLevel readCL;
@@ -432,28 +264,13 @@ public abstract class ResiliencyTestBase extends SharedClusterSparkIntegrationTe
}
}
- public static void fixDistributedSchemas(UpgradeableCluster cluster)
+ /**
+ * An interface that pulls a method from the Cassandra Storage Service Proxy
+ */
+ public interface SSProxy
{
- // These keyspaces are under replicated by default, so must be updated when doing a multi-node cluster;
- // else bootstrap will fail with 'Unable to find sufficient sources for streaming range <range> in keyspace <name>'
- for (String ks : Arrays.asList("system_auth", "system_traces"))
- {
- cluster.schemaChange("ALTER KEYSPACE " + ks +
- " WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': "
- + Math.min(cluster.size(), 3) + "}",
- true,
- cluster.getFirstRunningInstance());
- }
-
- // in real live repair is needed in this case, but in the test case it doesn't matter if the tables loose
- // anything, so ignoring repair to speed up the tests.
- }
+ Map<List<String>, List<String>> getRangeToEndpointWithPortMap(String keyspace);
- public static void waitForHealthyRing(UpgradeableCluster cluster)
- {
- for (IUpgradeableInstance inst : cluster)
- {
- ClusterUtils.awaitRingHealthy(inst);
- }
+ Map<List<String>, List<String>> getPendingRangeToEndpointWithPortMap(String keyspace);
}
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java
index 764c4a8..d92a737 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java
@@ -19,23 +19,21 @@
package org.apache.cassandra.analytics;
-import java.net.UnknownHostException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
-import java.util.stream.IntStream;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestInstance.Lifecycle;
import org.junit.jupiter.api.extension.ExtendWith;
+import com.datastax.driver.core.ResultSet;
import com.vdurmont.semver4j.Semver;
import io.vertx.junit5.VertxExtension;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CassandraBridgeFactory;
-import org.apache.cassandra.distributed.shared.JMXUtil;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.sidecar.testing.SharedClusterIntegrationTestBase;
import org.apache.spark.SparkConf;
@@ -45,11 +43,7 @@ import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructField;
-import relocated.shaded.com.datastax.driver.core.ResultSet;
-import static org.apache.cassandra.analytics.SparkTestUtils.defaultBulkReaderDataFrame;
-import static org.apache.cassandra.analytics.SparkTestUtils.defaultBulkWriterDataFrameWriter;
-import static org.apache.cassandra.analytics.SparkTestUtils.defaultSparkConf;
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -62,22 +56,26 @@ public abstract class SharedClusterSparkIntegrationTestBase extends SharedCluste
{
protected SparkConf sparkConf;
protected SparkSession sparkSession;
+ protected SparkTestUtils sparkTestUtils;
protected CassandraBridge bridge;
- public void validateWritesWithDriverResultSet(List<Row> sourceData, ResultSet queriedData,
- Function<relocated.shaded.com.datastax.driver.core.Row, String> rowFormatter)
+ public SharedClusterSparkIntegrationTestBase()
{
- Set<String> actualEntries = new HashSet<>();
- queriedData.forEach(row -> actualEntries.add(rowFormatter.apply(row)));
+ sparkTestUtils = SparkTestUtilsProvider.utils();
+ }
- // Number of entries in Cassandra must match the original datasource
- assertThat(actualEntries.size()).isEqualTo(sourceData.size());
+ @Override
+ protected void beforeTestStart()
+ {
+ super.beforeTestStart();
+ sparkTestUtils.initialize(cluster.delegate(), dnsResolver, server.actualPort());
+ }
- // remove from actual entries to make sure that the data read is the same as the data written
- Set<String> sourceEntries = sourceData.stream().map(this::getFormattedSourceEntry)
- .collect(Collectors.toSet());
- assertThat(actualEntries).as("All entries are expected to be read from database")
- .containsExactlyInAnyOrderElementsOf(sourceEntries);
+ @Override
+ protected void afterClusterShutdown()
+ {
+ super.afterClusterShutdown();
+ sparkTestUtils.tearDown();
}
/**
@@ -89,11 +87,8 @@ public abstract class SharedClusterSparkIntegrationTestBase extends SharedCluste
*/
protected DataFrameReader bulkReaderDataFrame(QualifiedName tableName)
{
- return defaultBulkReaderDataFrame(getOrCreateSparkConf(),
- getOrCreateSparkSession(),
- tableName,
- sidecarInstancesOption(),
- server.actualPort());
+ return sparkTestUtils.defaultBulkReaderDataFrame(getOrCreateSparkConf(), getOrCreateSparkSession(),
+ tableName);
}
/**
@@ -106,34 +101,14 @@ public abstract class SharedClusterSparkIntegrationTestBase extends SharedCluste
*/
protected DataFrameWriter<Row> bulkWriterDataFrameWriter(Dataset<Row> df, QualifiedName tableName)
{
- return defaultBulkWriterDataFrameWriter(df, tableName, sidecarInstancesOption(), server.actualPort());
- }
-
- /**
- * @return a comma-separated string with a list of all the hosts in the in-jvm dtest cluster
- */
- protected String sidecarInstancesOption()
- {
- return IntStream.rangeClosed(1, cluster.size())
- .mapToObj(i -> {
- String ipAddress = JMXUtil.getJmxHost(cluster.get(i).config());
- try
- {
- return dnsResolver.reverseResolve(ipAddress);
- }
- catch (UnknownHostException e)
- {
- return ipAddress;
- }
- })
- .collect(Collectors.joining(","));
+ return sparkTestUtils.defaultBulkWriterDataFrameWriter(df, tableName);
}
protected SparkConf getOrCreateSparkConf()
{
if (sparkConf == null)
{
- sparkConf = defaultSparkConf();
+ sparkConf = sparkTestUtils.defaultSparkConf();
}
return sparkConf;
}
@@ -160,7 +135,23 @@ public abstract class SharedClusterSparkIntegrationTestBase extends SharedCluste
return bridge;
}
- private String getFormattedSourceEntry(Row row)
+ public void validateWritesWithDriverResultSet(List<Row> sourceData, ResultSet queriedData,
+ Function<com.datastax.driver.core.Row, String> rowFormatter)
+ {
+ Set<String> actualEntries = new HashSet<>();
+ queriedData.forEach(row -> actualEntries.add(rowFormatter.apply(row)));
+
+ // Number of entries in Cassandra must match the original datasource
+ assertThat(actualEntries.size()).isEqualTo(sourceData.size());
+
+ // remove from actual entries to make sure that the data read is the same as the data written
+ Set<String> sourceEntries = sourceData.stream().map(this::formattedSourceEntry)
+ .collect(Collectors.toSet());
+ assertThat(actualEntries).as("All entries are expected to be read from database")
+ .containsExactlyInAnyOrderElementsOf(sourceEntries);
+ }
+
+ private String formattedSourceEntry(Row row)
{
StringBuilder sb = new StringBuilder();
for (int i = 0; i < row.size(); i++)
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkBulkWriterSimpleTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkBulkWriterSimpleTest.java
index 87e1c86..ebb6860 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkBulkWriterSimpleTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkBulkWriterSimpleTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -32,15 +31,9 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
-import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
@@ -101,25 +94,10 @@ class SparkBulkWriterSimpleTest extends SharedClusterSparkIntegrationTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected ClusterBuilderConfiguration testClusterConfiguration()
{
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(3)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(3, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- return clusterBuilder.start();
+ return super.testClusterConfiguration()
+ .nodesPerDc(3);
}
// Because the read part of the integration test job doesn't read ttl and timestamp columns, we need to remove them
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java
index 53806cb..46b70a9 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java
@@ -19,12 +19,19 @@
package org.apache.cassandra.analytics;
+import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.List;
+import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.shared.JMXUtil;
+import org.apache.cassandra.sidecar.common.dns.DnsResolver;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.KryoRegister;
import org.apache.cassandra.spark.bulkwriter.BulkSparkConf;
@@ -42,29 +49,45 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* Helper methods for Spark tests
*/
-public final class SparkTestUtils
+public class SparkTestUtils
{
- private SparkTestUtils()
+ protected ICluster<? extends IInstance> cluster;
+ private DnsResolver dnsResolver;
+ private int sidecarPort;
+
+ /**
+ * Runs any initialization code required for the tests
+ *
+ * @param cluster the cassandra cluster
+ * @param dnsResolver the DNS resolver used to lookup replicas
+ * @param sidecarPort the port where Sidecar is running
+ */
+ public void initialize(ICluster<? extends IInstance> cluster, DnsResolver dnsResolver, int sidecarPort)
+ {
+ this.cluster = Objects.requireNonNull(cluster, "cluster is required");
+ this.dnsResolver = Objects.requireNonNull(dnsResolver, "dnsResolver is required");
+ this.sidecarPort = sidecarPort;
+ }
+
+ /**
+ * Runs any tear down code required for tests, currently empty
+ */
+ public void tearDown()
{
- throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated");
}
/**
* Returns a {@link DataFrameReader} with default options for performing a bulk read test, including
* required parameters.
*
- * @param sparkConf the spark configuration to use
- * @param spark the spark session to use
- * @param tableName the qualified name of the table
- * @param sidecarInstancesOption the comma-separated list of sidecar instances
- * @param sidecarPort the sidecar port
+ * @param sparkConf the spark configuration to use
+ * @param spark the spark session to use
+ * @param tableName the qualified name of the table
* @return a {@link DataFrameReader} with default options for performing a bulk read test
*/
- public static DataFrameReader defaultBulkReaderDataFrame(SparkConf sparkConf,
- SparkSession spark,
- QualifiedName tableName,
- String sidecarInstancesOption,
- int sidecarPort)
+ public DataFrameReader defaultBulkReaderDataFrame(SparkConf sparkConf,
+ SparkSession spark,
+ QualifiedName tableName)
{
SQLContext sql = spark.sqlContext();
SparkContext sc = spark.sparkContext();
@@ -74,7 +97,7 @@ public final class SparkTestUtils
int numCores = coresPerExecutor * numExecutors;
return sql.read().format("org.apache.cassandra.spark.sparksql.CassandraDataSource")
- .option("sidecar_instances", sidecarInstancesOption)
+ .option("sidecar_instances", sidecarInstancesOption(cluster, dnsResolver))
.option("keyspace", tableName.keyspace()) // unquoted
.option("table", tableName.table()) // unquoted
.option("DC", "datacenter1")
@@ -94,20 +117,16 @@ public final class SparkTestUtils
* Returns a {@link DataFrameWriter<Row>} with default options for performing a bulk write test, including
* required parameters.
*
- * @param df the source data frame
- * @param tableName the qualified name of the table
- * @param sidecarInstancesOption the comma-separated list of sidecar instances
- * @param sidecarPort the sidecar port
+ * @param df the source data frame
+ * @param tableName the qualified name of the table
* @return a {@link DataFrameWriter<Row>} with default options for performing a bulk write test
*/
- public static DataFrameWriter<Row> defaultBulkWriterDataFrameWriter(Dataset<Row> df,
- QualifiedName tableName,
- String sidecarInstancesOption,
- int sidecarPort)
+ public DataFrameWriter<Row> defaultBulkWriterDataFrameWriter(Dataset<Row> df,
+ QualifiedName tableName)
{
return df.write()
.format("org.apache.cassandra.spark.sparksql.CassandraDataSink")
- .option("sidecar_instances", sidecarInstancesOption)
+ .option("sidecar_instances", sidecarInstancesOption(cluster, dnsResolver))
.option("keyspace", tableName.keyspace())
.option("table", tableName.table())
.option("local_dc", "datacenter1")
@@ -117,7 +136,7 @@ public final class SparkTestUtils
.mode("append");
}
- public static SparkConf defaultSparkConf()
+ public SparkConf defaultSparkConf()
{
SparkConf sparkConf = new SparkConf()
.setAppName("Integration test Spark Cassandra Bulk Analytics Job")
@@ -134,7 +153,7 @@ public final class SparkTestUtils
return sparkConf;
}
- public static void validateWrites(List<Row> sourceData, Object[][] queriedData)
+ public void validateWrites(List<Row> sourceData, Object[][] queriedData)
{
// build a set of entries read from Cassandra into a set
Set<String> actualEntries = Arrays.stream(queriedData)
@@ -161,4 +180,24 @@ public final class SparkTestUtils
assertThat(actualEntries).as("All entries are expected to be read from database")
.isEmpty();
}
+
+ /**
+ * @return a comma-separated string with a list of all the hosts in the in-jvm dtest cluster
+ */
+ protected String sidecarInstancesOption(ICluster<? extends IInstance> cluster, DnsResolver dnsResolver)
+ {
+ return IntStream.rangeClosed(1, cluster.size())
+ .mapToObj(i -> {
+ String ipAddress = JMXUtil.getJmxHost(cluster.get(i).config());
+ try
+ {
+ return dnsResolver.reverseResolve(ipAddress);
+ }
+ catch (UnknownHostException e)
+ {
+ return ipAddress;
+ }
+ })
+ .collect(Collectors.joining(","));
+ }
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtilsProvider.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtilsProvider.java
new file mode 100644
index 0000000..fed5842
--- /dev/null
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtilsProvider.java
@@ -0,0 +1,56 @@
+/*
+ * 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 org.apache.cassandra.analytics;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A class that provides {@link SparkTestUtils}
+ */
+public final class SparkTestUtilsProvider
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(SparkTestUtilsProvider.class);
+
+ private SparkTestUtilsProvider()
+ {
+ throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated");
+ }
+
+ public static SparkTestUtils utils()
+ {
+ String className = System.getProperty("cassandra.analytics.spark.test_utils");
+
+ if (className != null)
+ {
+ try
+ {
+ LOGGER.info("Instantiating SparkTestUtils with class name {}", className);
+ return (SparkTestUtils) Class.forName(className).getDeclaredConstructor().newInstance();
+ }
+ catch (Exception e)
+ {
+ LOGGER.warn("Unable to initialize class {}", className, e);
+ }
+ }
+
+ return new SparkTestUtils();
+ }
+}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampIntegrationTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampIntegrationTest.java
index 7f9f5bf..98b96d6 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampIntegrationTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampIntegrationTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.time.Instant;
import java.util.Arrays;
import java.util.Comparator;
@@ -30,23 +29,15 @@ import java.util.stream.Collectors;
import org.junit.jupiter.api.Test;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICoordinator;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.TimestampOption;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
-import static org.apache.cassandra.analytics.ResiliencyTestBase.fixDistributedSchemas;
-import static org.apache.cassandra.analytics.ResiliencyTestBase.waitForHealthyRing;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
import static org.apache.cassandra.testing.TestUtils.DC1_RF1;
import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
@@ -94,29 +85,10 @@ class TimestampIntegrationTest extends SharedClusterSparkIntegrationTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected ClusterBuilderConfiguration testClusterConfiguration()
{
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(3)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(3, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- UpgradeableCluster cluster = clusterBuilder.start();
-
- waitForHealthyRing(cluster);
- fixDistributedSchemas(cluster);
- return cluster;
+ return super.testClusterConfiguration()
+ .nodesPerDc(3);
}
void validateWrites(QualifiedName tableName, List<Row> sourceData)
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/WriterDigestIntegrationTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/WriterDigestIntegrationTest.java
index 8589a19..ee9b7ed 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/WriterDigestIntegrationTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/WriterDigestIntegrationTest.java
@@ -19,25 +19,17 @@
package org.apache.cassandra.analytics;
-import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import org.junit.jupiter.api.Test;
-import com.vdurmont.semver4j.Semver;
-import org.apache.cassandra.distributed.UpgradeableCluster;
-import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.Versions;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
-import static org.apache.cassandra.analytics.SparkTestUtils.validateWrites;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
import static org.apache.cassandra.testing.TestUtils.DC1_RF1;
import static org.apache.cassandra.testing.TestUtils.ROW_COUNT;
@@ -60,7 +52,7 @@ class WriterDigestIntegrationTest extends SharedClusterSparkIntegrationTestBase
void testDefaultDigest()
{
bulkWriterDataFrameWriter(df, DEFAULT_DIGEST_TABLE).save();
- validateWrites(df.collectAsList(), queryAllData(DEFAULT_DIGEST_TABLE));
+ sparkTestUtils.validateWrites(df.collectAsList(), queryAllData(DEFAULT_DIGEST_TABLE));
}
@Test
@@ -69,7 +61,7 @@ class WriterDigestIntegrationTest extends SharedClusterSparkIntegrationTestBase
SparkSession spark = getOrCreateSparkSession();
Dataset<Row> df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT);
bulkWriterDataFrameWriter(df, MD5_DIGEST_TABLE).option(WriterOptions.DIGEST.name(), "MD5").save();
- validateWrites(df.collectAsList(), queryAllData(MD5_DIGEST_TABLE));
+ sparkTestUtils.validateWrites(df.collectAsList(), queryAllData(MD5_DIGEST_TABLE));
}
@Test
@@ -84,33 +76,11 @@ class WriterDigestIntegrationTest extends SharedClusterSparkIntegrationTestBase
@Override
protected void beforeTestStart()
{
+ super.beforeTestStart();
SparkSession spark = getOrCreateSparkSession();
df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT);
}
- @Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
- {
- // spin up a C* cluster using the in-jvm dtest
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(), Semver.SemverType.LOOSE));
-
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(1)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDCs(1)
- .withDataDirCount(1)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
- TokenSupplier tokenSupplier = TokenSupplier.evenlyDistributedTokens(1, clusterBuilder.getTokenCount());
- clusterBuilder.withTokenSupplier(tokenSupplier);
- UpgradeableCluster cluster = clusterBuilder.createWithoutStarting();
- cluster.startup();
- return cluster;
- }
-
@Override
protected void initializeSchemaForTest()
{
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/data/ClearSnapshotTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/data/ClearSnapshotTest.java
index 2b7b0a1..261502a 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/data/ClearSnapshotTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/data/ClearSnapshotTest.java
@@ -31,18 +31,10 @@ import java.util.stream.Stream;
import org.junit.jupiter.api.Test;
-import com.vdurmont.semver4j.Semver;
import org.apache.cassandra.analytics.SharedClusterSparkIntegrationTestBase;
-import org.apache.cassandra.config.CassandraRelevantProperties;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.shared.Uninterruptibles;
-import org.apache.cassandra.distributed.shared.Versions;
-import org.apache.cassandra.distributed.shared.WithProperties;
-import org.apache.cassandra.sidecar.testing.JvmDTestSharedClassesPredicate;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.testing.TestVersion;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.Row;
@@ -52,7 +44,6 @@ import static org.assertj.core.api.Assertions.assertThat;
class ClearSnapshotTest extends SharedClusterSparkIntegrationTestBase
{
- private static final WithProperties properties = new WithProperties();
static final QualifiedName TABLE_NAME_FOR_TTL_CLEAR_SNAPSHOT_STRATEGY
= new QualifiedName(TEST_KEYSPACE, "test_ttl_clear_snapshot_strategy");
static final QualifiedName TABLE_NAME_FOR_NO_OP_CLEAR_SNAPSHOT_STRATEGY
@@ -68,7 +59,7 @@ class ClearSnapshotTest extends SharedClusterSparkIntegrationTestBase
List<Row> rows = readDf.load().collectAsList();
assertThat(rows.size()).isEqualTo(8);
- String[] dataDirs = (String[]) cluster.getFirstRunningInstance()
+ String[] dataDirs = (String[]) cluster.get(1)
.config()
.getParams()
.get("data_file_directories");
@@ -96,7 +87,7 @@ class ClearSnapshotTest extends SharedClusterSparkIntegrationTestBase
List<Row> rows = readDf.load().collectAsList();
assertThat(rows.size()).isEqualTo(8);
- String[] dataDirs = (String[]) cluster.getFirstRunningInstance()
+ String[] dataDirs = (String[]) cluster.get(1)
.config()
.getParams()
.get("data_file_directories");
@@ -124,33 +115,19 @@ class ClearSnapshotTest extends SharedClusterSparkIntegrationTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected void beforeClusterProvisioning()
{
- properties.set(CassandraRelevantProperties.SNAPSHOT_CLEANUP_INITIAL_DELAY_SECONDS, 0);
- properties.set(CassandraRelevantProperties.SNAPSHOT_CLEANUP_PERIOD_SECONDS, 1);
- properties.set(CassandraRelevantProperties.SNAPSHOT_MIN_ALLOWED_TTL_SECONDS, 5);
-
- Versions versions = Versions.find();
- Versions.Version requestedVersion = versions.getLatest(new Semver(testVersion.version(),
- Semver.SemverType.LOOSE));
- UpgradeableCluster.Builder clusterBuilder =
- UpgradeableCluster.build(1)
- .withDynamicPortAllocation(true)
- .withVersion(requestedVersion)
- .withDataDirCount(1)
- .withDCs(1)
- .withSharedClasses(JvmDTestSharedClassesPredicate.INSTANCE)
- .withConfig(config -> config.with(Feature.NATIVE_PROTOCOL)
- .with(Feature.GOSSIP)
- .with(Feature.JMX));
-
- return clusterBuilder.start();
+ System.setProperty("cassandra.snapshot.ttl_cleanup_initial_delay_seconds", "0");
+ System.setProperty("cassandra.snapshot.ttl_cleanup_period_seconds", "1");
+ System.setProperty("cassandra.snapshot.min_allowed_ttl_seconds", "5");
}
@Override
protected void afterClusterShutdown()
{
- properties.close();
+ System.clearProperty("cassandra.snapshot.ttl_cleanup_initial_delay_seconds");
+ System.clearProperty("cassandra.snapshot.ttl_cleanup_period_seconds");
+ System.clearProperty("cassandra.snapshot.min_allowed_ttl_seconds");
}
@Override
@@ -170,7 +147,7 @@ class ClearSnapshotTest extends SharedClusterSparkIntegrationTestBase
{
String value = DATASET.get(i);
String query = String.format("INSERT INTO %s (c1, c2) VALUES (%d, '%s');", tableName, i, value);
- cluster.getFirstRunningInstance()
+ cluster.get(1)
.coordinator()
.execute(query, ConsistencyLevel.ALL);
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterFailureTest.java
index b05f359..dd3c42c 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterFailureTest.java
@@ -34,7 +34,7 @@ import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -71,7 +71,6 @@ class JoiningDoubleClusterFailureTest extends JoiningSingleNodeTest
/**
* ByteBuddy helper for doubling cluster size failure scenario
*/
- @Shared
public static class BBHelperDoubleClusterSizeFailure
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(5);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterTest.java
index 7adf13a..0ac7b2a 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningDoubleClusterTest.java
@@ -33,7 +33,7 @@ import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -70,7 +70,6 @@ class JoiningDoubleClusterTest extends JoiningSingleNodeTest
/**
* ByteBuddy helper for doubling cluster size
*/
- @Shared
public static class BBHelperDoubleClusterSize
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(5);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCFailureTest.java
index 44cfced..2a7cdbe 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCFailureTest.java
@@ -34,7 +34,7 @@ import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -68,7 +68,6 @@ class JoiningMultiDCFailureTest extends JoiningMultiDCTest
/**
* ByteBuddy helper for multiple joining nodes failure scenario in multiDC
*/
- @Shared
public static class BBHelperMultiDCFailure
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(6);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedFailureTest.java
index 0259a77..a05b16d 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedFailureTest.java
@@ -34,7 +34,7 @@ import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -68,7 +68,6 @@ class JoiningMultiDCSingleReplicatedFailureTest extends JoiningMultiDCSingleRepl
/**
* ByteBuddy helper for multiple joining nodes failure
*/
- @Shared
public static class BBHelperMultiDCFailure
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedTest.java
index 988ccb1..172a443 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCSingleReplicatedTest.java
@@ -34,7 +34,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -81,7 +81,6 @@ class JoiningMultiDCSingleReplicatedTest extends JoiningMultiDCTest
/**
* ByteBuddy helper for multiple joining nodes
*/
- @Shared
public static class BBHelperMultiDC
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCTest.java
index 4f27add..875a163 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultiDCTest.java
@@ -39,7 +39,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static com.datastax.driver.core.ConsistencyLevel.ALL;
import static com.datastax.driver.core.ConsistencyLevel.EACH_QUORUM;
@@ -110,7 +110,6 @@ class JoiningMultiDCTest extends JoiningTestBase
/**
* ByteBuddy helper for multiple joining nodes
*/
- @Shared
public static class BBHelperMultiDC
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(6);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesFailureTest.java
index a570d4c..6700354 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesFailureTest.java
@@ -34,7 +34,7 @@ import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -71,7 +71,6 @@ class JoiningMultipleNodesFailureTest extends JoiningMultipleNodesTest
/**
* ByteBuddy helper for multiple joining nodes failure scenario
*/
- @Shared
public static class BBHelperMultipleJoiningNodesFailure
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesTest.java
index f2f855f..00173be 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningMultipleNodesTest.java
@@ -33,7 +33,7 @@ import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -69,7 +69,6 @@ class JoiningMultipleNodesTest extends JoiningSingleNodeTest
/**
* ByteBuddy helper for multiple joining nodes
*/
- @Shared
public static class BBHelperMultipleJoiningNodes
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeFailureTest.java
index bea9a28..036cad9 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeFailureTest.java
@@ -33,7 +33,7 @@ import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -67,7 +67,6 @@ class JoiningSingleNodeFailureTest extends JoiningSingleNodeTest
/**
* ByteBuddy helper for a single joining node failure case
*/
- @Shared
public static class BBHelperSingleJoiningNodeFailure
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeTest.java
index a345ecc..403dbb7 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningSingleNodeTest.java
@@ -39,7 +39,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static com.datastax.driver.core.ConsistencyLevel.ALL;
import static com.datastax.driver.core.ConsistencyLevel.ONE;
@@ -104,7 +104,6 @@ class JoiningSingleNodeTest extends JoiningTestBase
/**
* ByteBuddy helper for a single joining node
*/
- @Shared
public static class BBHelperSingleJoiningNode
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningTestBase.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningTestBase.java
index 16ac634..3a534c6 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningTestBase.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/expansion/JoiningTestBase.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.analytics.expansion;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -33,13 +32,13 @@ import org.junit.jupiter.params.provider.Arguments;
import org.apache.cassandra.analytics.DataGenerationUtils;
import org.apache.cassandra.analytics.ResiliencyTestBase;
import org.apache.cassandra.analytics.TestUninterruptibles;
-import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.testing.utils.ClusterUtils;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstance;
-import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
+import org.apache.cassandra.testing.IClusterExtension;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
@@ -67,6 +66,7 @@ abstract class JoiningTestBase extends ResiliencyTestBase
@Override
protected void beforeTestStart()
{
+ super.beforeTestStart();
SparkSession spark = getOrCreateSparkSession();
// Generate some artificial data for the test
df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT);
@@ -75,16 +75,12 @@ abstract class JoiningTestBase extends ResiliencyTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected void afterClusterProvisioned()
{
ClusterBuilderConfiguration configuration = testClusterConfiguration();
- UpgradeableCluster provisionedCluster = clusterBuilder(configuration, testVersion);
-
- newInstances = addNewInstances(provisionedCluster, configuration.newNodesPerDc, configuration.dcCount);
+ newInstances = addNewInstances(cluster, configuration.newNodesPerDc, configuration.dcCount);
TestUninterruptibles.awaitUninterruptiblyOrThrow(transitioningStateStart(), 2, TimeUnit.MINUTES);
- newInstances.forEach(instance -> ClusterUtils.awaitRingState(instance, instance, "Joining"));
-
- return provisionedCluster;
+ newInstances.forEach(instance -> cluster.awaitRingState(instance, instance, "Joining"));
}
protected void completeTransitionsAndValidateWrites(CountDownLatch transitionalStateEnd,
@@ -124,12 +120,7 @@ abstract class JoiningTestBase extends ResiliencyTestBase
*/
protected abstract CountDownLatch transitioningStateStart();
- /**
- * @return the configuration for the test cluster
- */
- protected abstract ClusterBuilderConfiguration testClusterConfiguration();
-
- private static List<IInstance> addNewInstances(UpgradeableCluster cluster, int newNodesPerDc, int numDcs)
+ private static List<IInstance> addNewInstances(IClusterExtension<? extends IInstance> cluster, int newNodesPerDc, int numDcs)
{
List<IInstance> newInstances = new ArrayList<>();
// Go over new nodes and add them once for each DC
@@ -139,16 +130,15 @@ abstract class JoiningTestBase extends ResiliencyTestBase
for (int dc = 1; dc <= numDcs; dc++)
{
IInstance dcNode = cluster.get(dcNodeIdx++);
- IInstance newInstance = ClusterUtils.addInstance(cluster,
- dcNode.config().localDatacenter(),
- dcNode.config().localRack(),
- inst -> {
- inst.set("auto_bootstrap", true);
- inst.with(Feature.GOSSIP,
- Feature.JMX,
- Feature.NATIVE_PROTOCOL);
- });
- new Thread(() -> newInstance.startup(cluster)).start();
+ IInstance newInstance = cluster.addInstance(dcNode.config().localDatacenter(),
+ dcNode.config().localRack(),
+ inst -> {
+ inst.set("auto_bootstrap", true);
+ inst.with(Feature.GOSSIP,
+ Feature.JMX,
+ Feature.NATIVE_PROTOCOL);
+ });
+ new Thread(() -> newInstance.startup(cluster.delegate())).start();
newInstances.add(newInstance);
}
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementFailureTest.java
index a8871c9..d5d00f3 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementFailureTest.java
@@ -34,7 +34,7 @@ import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -66,7 +66,6 @@ class NodeMovementFailureTest extends NodeMovementTest
/**
* ByteBuddy Helper for a single moving node simulating a move failure
*/
- @Shared
public static class BBHelperMovingNodeFailure
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCFailureTest.java
index 3309383..3a0441d 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCFailureTest.java
@@ -34,7 +34,7 @@ import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -67,7 +67,6 @@ class NodeMovementMultiDCFailureTest extends NodeMovementMultiDCTest
/**
* ByteBuddy Helper for a single moving node
*/
- @Shared
public static class BBHelperMultiDCMovingNodeFailure
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCTest.java
index b7a294a..5b86a2a 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementMultiDCTest.java
@@ -39,7 +39,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static com.datastax.driver.core.ConsistencyLevel.ALL;
import static com.datastax.driver.core.ConsistencyLevel.EACH_QUORUM;
@@ -109,7 +109,6 @@ class NodeMovementMultiDCTest extends NodeMovementTestBase
/**
* ByteBuddy Helper for a single moving node
*/
- @Shared
public static class BBHelperMovingNodeMultiDC
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTest.java
index 0e58a76..8fbd319 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTest.java
@@ -39,7 +39,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static com.datastax.driver.core.ConsistencyLevel.ALL;
import static com.datastax.driver.core.ConsistencyLevel.ONE;
@@ -104,7 +104,6 @@ class NodeMovementTest extends NodeMovementTestBase
/**
* ByteBuddy Helper for a single moving node
*/
- @Shared
public static class BBHelperMovingNode
{
static final CountDownLatch transitioningStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTestBase.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTestBase.java
index f336ffd..186717d 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTestBase.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/movement/NodeMovementTestBase.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.analytics.movement;
-import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
@@ -32,13 +31,12 @@ import org.junit.jupiter.params.provider.Arguments;
import org.apache.cassandra.analytics.DataGenerationUtils;
import org.apache.cassandra.analytics.ResiliencyTestBase;
import org.apache.cassandra.analytics.TestUninterruptibles;
-import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.testing.utils.ClusterUtils;
+import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInstance;
-import org.apache.cassandra.distributed.impl.AbstractCluster;
-import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
@@ -69,6 +67,7 @@ abstract class NodeMovementTestBase extends ResiliencyTestBase
@Override
protected void beforeTestStart()
{
+ super.beforeTestStart();
SparkSession spark = getOrCreateSparkSession();
// Generate some artificial data for the test
df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT);
@@ -77,25 +76,21 @@ abstract class NodeMovementTestBase extends ResiliencyTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected void afterClusterProvisioned()
{
ClusterBuilderConfiguration configuration = testClusterConfiguration();
- UpgradeableCluster provisionedCluster = clusterBuilder(configuration, testVersion);
-
int movingNodeIndex = configuration.dcCount > 1 ? MULTI_DC_MOVING_NODE_IDX : SINGLE_DC_MOVING_NODE_IDX;
- movingNode = provisionedCluster.get(movingNodeIndex);
+ movingNode = cluster.get(movingNodeIndex);
- IInstance seed = provisionedCluster.get(1);
+ IInstance seed = cluster.get(1);
new Thread(() -> {
- long moveTarget = calculateMoveTargetToken(provisionedCluster, configuration.dcCount);
+ long moveTarget = calculateMoveTargetToken(cluster, configuration.dcCount);
movingNode.nodetoolResult("move", "--", Long.toString(moveTarget)).asserts().success();
}).start();
// Wait until nodes have reached expected state
TestUninterruptibles.awaitUninterruptiblyOrThrow(transitioningStateStart(), 2, TimeUnit.MINUTES);
- ClusterUtils.awaitRingState(seed, movingNode, "Moving");
-
- return provisionedCluster;
+ cluster.awaitRingState(seed, movingNode, "Moving");
}
/**
@@ -114,7 +109,7 @@ abstract class NodeMovementTestBase extends ResiliencyTestBase
// It is only in successful MOVE operation that we validate that the node has reached NORMAL state
if (!expectFailure)
{
- ClusterUtils.awaitRingState(cluster.get(1), movingNode, "Normal");
+ cluster.awaitRingState(cluster.get(1), movingNode, "Normal");
}
testInputs.forEach(arguments -> {
@@ -143,12 +138,7 @@ abstract class NodeMovementTestBase extends ResiliencyTestBase
}
}
- /**
- * @return the configuration for the test cluster
- */
- protected abstract ClusterBuilderConfiguration testClusterConfiguration();
-
- static long calculateMoveTargetToken(AbstractCluster<? extends IInstance> cluster, int dcCount)
+ static long calculateMoveTargetToken(ICluster<? extends IInstance> cluster, int dcCount)
{
IInstance seed = cluster.get(1);
// The target token to move the node to is calculated by adding an offset to the seed node token which
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementFailureTest.java
index 62d04e3..305f53f 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementFailureTest.java
@@ -33,7 +33,7 @@ import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -66,7 +66,6 @@ class HostReplacementFailureTest extends HostReplacementTest
/**
* ByteBuddy helper for a single node replacement
*/
- @Shared
public static class BBHelperReplacementsNodeFailure
{
// Additional latch used here to sequentially start the 2 new nodes to isolate the loading
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCFailureTest.java
index 1628ef5..554e648 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCFailureTest.java
@@ -39,7 +39,7 @@ import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static com.datastax.driver.core.ConsistencyLevel.LOCAL_QUORUM;
import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -99,7 +99,6 @@ class HostReplacementMultiDCFailureTest extends HostReplacementTestBase
/**
* ByteBuddy helper for multi DC node replacement failure
*/
- @Shared
public static class BBHelperReplacementFailureMultiDC
{
// Additional latch used here to sequentially start the 2 new nodes to isolate the loading
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCInsufficientReplicasTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCInsufficientReplicasTest.java
index 81a9689..8422493 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCInsufficientReplicasTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCInsufficientReplicasTest.java
@@ -38,8 +38,8 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.cassandra.testing.TestUtils;
-import org.apache.cassandra.utils.Shared;
import static com.datastax.driver.core.ConsistencyLevel.EACH_QUORUM;
import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -121,7 +121,6 @@ class HostReplacementMultiDCInsufficientReplicasTest extends HostReplacementTest
/**
* ByteBuddy helper for multi DC node replacement failure resulting in insufficient nodes
*/
- @Shared
public static class BBHelperNodeReplacementMultiDCInsufficientReplicas
{
// Additional latch used here to sequentially start the 2 new nodes to isolate the loading
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCTest.java
index 4cbb235..7d778d4 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementMultiDCTest.java
@@ -40,7 +40,7 @@ import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static com.datastax.driver.core.ConsistencyLevel.ALL;
import static com.datastax.driver.core.ConsistencyLevel.EACH_QUORUM;
@@ -114,7 +114,6 @@ class HostReplacementMultiDCTest extends HostReplacementTestBase
/**
* ByteBuddy helper for a multi DC node replacement that succeeds
*/
- @Shared
public static class BBHelperNodeReplacementMultiDC
{
// Additional latch used here to sequentially start the 2 new nodes to isolate the loading
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTest.java
index 46990b9..4d83121 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTest.java
@@ -38,7 +38,7 @@ import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
@@ -95,7 +95,6 @@ class HostReplacementTest extends HostReplacementTestBase
/**
* ByteBuddy helper for a single node replacement
*/
- @Shared
public static class BBHelperReplacementsNode
{
// Additional latch used here to sequentially start the 2 new nodes to isolate the loading
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTestBase.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTestBase.java
index 7ca4add..2d031cc 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTestBase.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/replacement/HostReplacementTestBase.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.analytics.replacement;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -37,17 +36,13 @@ import org.junit.jupiter.params.provider.Arguments;
import org.apache.cassandra.analytics.DataGenerationUtils;
import org.apache.cassandra.analytics.ResiliencyTestBase;
-import org.apache.cassandra.config.CassandraRelevantProperties;
-import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.distributed.api.IInstanceConfig;
-import org.apache.cassandra.distributed.impl.AbstractCluster;
-import org.apache.cassandra.distributed.impl.InstanceConfig;
-import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.IClusterExtension;
+import org.apache.cassandra.testing.utils.ClusterUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
@@ -67,18 +62,15 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
List<String> removedNodeAddresses;
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected void afterClusterProvisioned()
{
- ClusterBuilderConfiguration configuration = testClusterConfiguration();
- UpgradeableCluster provisionedCluster = clusterBuilder(configuration, testVersion);
+ assertThat(additionalNodesToStop()).isLessThan(cluster.size() - 1);
- assertThat(additionalNodesToStop()).isLessThan(provisionedCluster.size() - 1);
-
- IInstance seed = provisionedCluster.get(1);
+ IInstance seed = cluster.get(1);
List<ClusterUtils.RingInstanceDetails> ring = ClusterUtils.ring(seed);
// Remove the last node
- List<IInstance> nodesToRemove = Collections.singletonList(provisionedCluster.get(provisionedCluster.size()));
+ List<IInstance> nodesToRemove = Collections.singletonList(cluster.get(cluster.size()));
removedNodeAddresses = nodesToRemove.stream()
.map(n -> n.config()
.broadcastAddress()
@@ -94,9 +86,9 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
List<IInstance> additionalRemovalNodes = new ArrayList<>();
for (int i = 1; i <= additionalNodesToStop(); i++)
{
- additionalRemovalNodes.add(provisionedCluster.get(provisionedCluster.size() - i));
+ additionalRemovalNodes.add(cluster.get(cluster.size() - i));
}
- newNodes = startReplacementNodes(nodeStart(), provisionedCluster, nodesToRemove);
+ newNodes = startReplacementNodes(nodeStart(), cluster, nodesToRemove);
stopNodes(seed, additionalRemovalNodes);
// Wait until replacement nodes are in JOINING state
@@ -106,8 +98,8 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
// Verify state of replacement nodes
for (IInstance newInstance : newNodes)
{
- ClusterUtils.awaitRingState(newInstance, newInstance, "Joining");
- ClusterUtils.awaitGossipStatus(newInstance, newInstance, "BOOT_REPLACE");
+ cluster.awaitRingState(newInstance, newInstance, "Joining");
+ cluster.awaitGossipStatus(newInstance, newInstance, "BOOT_REPLACE");
String newAddress = newInstance.config().broadcastAddress().getAddress().getHostAddress();
Optional<ClusterUtils.RingInstanceDetails> replacementInstance = getMatchingInstanceFromRing(newInstance, newAddress);
@@ -115,8 +107,6 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
// Verify that replacement node tokens match the removed nodes
assertThat(removedNodeTokens).contains(replacementInstance.get().getToken());
}
-
- return provisionedCluster;
}
protected void completeTransitionsAndValidateWrites(CountDownLatch transitionalStateEnd,
@@ -135,7 +125,7 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
// It is only in successful REPLACE operation that we validate that the node has reached NORMAL state
if (!expectFailure)
{
- ClusterUtils.awaitRingState(newNodes.get(0), newNodes.get(0), "Normal");
+ cluster.awaitRingState(newNodes.get(0), newNodes.get(0), "Normal");
// Validate if data was written to the new transitioning nodes only when bootstrap succeeded
testInputs.forEach(arguments -> {
@@ -176,6 +166,7 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
@Override
protected void beforeTestStart()
{
+ super.beforeTestStart();
SparkSession spark = getOrCreateSparkSession();
// Generate some artificial data for the test
df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT);
@@ -191,11 +182,6 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
return 0;
}
- /**
- * @return the configuration for the test cluster
- */
- protected abstract ClusterBuilderConfiguration testClusterConfiguration();
-
/**
* @return a latch that will wait until the node starts
*/
@@ -209,7 +195,7 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
);
}
- public static <I extends IInstance> I addInstanceLocal(AbstractCluster<I> cluster,
+ public static <I extends IInstance> I addInstanceLocal(IClusterExtension<I> cluster,
String dc,
String rack,
Consumer<IInstanceConfig> fn,
@@ -217,14 +203,14 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
{
Objects.requireNonNull(dc, "dc");
Objects.requireNonNull(rack, "rack");
- InstanceConfig config = cluster.newInstanceConfig();
+ IInstanceConfig config = cluster.newInstanceConfig();
config.set("storage_port", remPort);
config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
fn.accept(config);
return cluster.bootstrap(config);
}
- private List<IInstance> startReplacementNodes(CountDownLatch nodeStart, UpgradeableCluster cluster,
+ private List<IInstance> startReplacementNodes(CountDownLatch nodeStart, IClusterExtension<?> cluster,
List<IInstance> nodesToRemove)
{
List<IInstance> newNodes = new ArrayList<>();
@@ -252,9 +238,8 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
new Thread(() -> ClusterUtils.start(replacement, (properties) -> {
replacement.config().set("storage_port", remPort);
- properties.set(CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK, true);
- properties.set(CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS,
- TimeUnit.SECONDS.toMillis(10L));
+ properties.with("cassandra.skip_schema_check", "true");
+ properties.with("cassandra.schema_delay_ms", String.valueOf(TimeUnit.SECONDS.toMillis(10L)));
properties.with("cassandra.broadcast_interval_ms",
Long.toString(TimeUnit.SECONDS.toMillis(30L)));
properties.with("cassandra.ring_delay_ms",
@@ -274,7 +259,7 @@ abstract class HostReplacementTestBase extends ResiliencyTestBase
{
for (IInstance node : nodesToRemove)
{
- ClusterUtils.stopUnchecked(node);
+ cluster.stopUnchecked(node);
String remAddress = node.config().broadcastAddress().getAddress().getHostAddress();
List<ClusterUtils.RingInstanceDetails> ring = ClusterUtils.ring(seed);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingHalfTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingHalfTest.java
index ffb5f70..8cbe1b8 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingHalfTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingHalfTest.java
@@ -36,7 +36,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -95,7 +95,6 @@ class LeavingHalfTest extends LeavingTestBase
/**
* ByteBuddy helper for shrinking cluster by half its size
*/
- @Shared
public static class BBHelperHalveClusterSize
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(3);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCFailureTest.java
index 186810a..63c43d6 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCFailureTest.java
@@ -37,7 +37,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -104,7 +104,6 @@ class LeavingMultiDCFailureTest extends LeavingTestBase
/**
* ByteBuddy helper for multiple leaving nodes multi-DC failure scenario
*/
- @Shared
public static class BBHelperLeavingNodesMultiDCFailure
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterFailureTest.java
index 63262ed..1b39ee6 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterFailureTest.java
@@ -37,7 +37,7 @@ import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -106,7 +106,6 @@ class LeavingMultiDCHalveClusterFailureTest extends LeavingTestBase
/**
* ByteBuddy helper for halve cluster size with multi-DC failure scenario
*/
- @Shared
public static class BBHelperHalveClusterMultiDCFailure
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(6);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterTest.java
index a50771e..91a6290 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCHalveClusterTest.java
@@ -41,7 +41,7 @@ import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -109,7 +109,6 @@ class LeavingMultiDCHalveClusterTest extends LeavingTestBase
/**
* ByteBuddy helper for halve cluster size with multi-DC
*/
- @Shared
public static class BBHelperHalveClusterMultiDC
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(6);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCTest.java
index 88c63a4..6f039f3 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultiDCTest.java
@@ -38,7 +38,7 @@ import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -100,7 +100,6 @@ class LeavingMultiDCTest extends LeavingTestBase
/**
* ByteBuddy helper for multiple leaving nodes multi-DC
*/
- @Shared
public static class BBHelperLeavingNodesMultiDC
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleFailureTest.java
index 55dddd6..2921ea4 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleFailureTest.java
@@ -36,7 +36,7 @@ import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -100,7 +100,6 @@ class LeavingMultipleFailureTest extends LeavingTestBase
/**
* ByteBuddy helper for multiple leaving nodes failure scenario
*/
- @Shared
public static class BBHelperMultipleLeavingNodesFailure
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleTest.java
index 9feacca..3d47d32 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingMultipleTest.java
@@ -36,7 +36,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -95,7 +95,6 @@ class LeavingMultipleTest extends LeavingTestBase
/**
* ByteBuddy helper for multiple leaving nodes
*/
- @Shared
public static class BBHelperMultipleLeavingNodes
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(2);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleFailureTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleFailureTest.java
index 8d96cc4..08f1cf2 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleFailureTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleFailureTest.java
@@ -36,7 +36,7 @@ import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -100,7 +100,6 @@ class LeavingSingleFailureTest extends LeavingTestBase
/**
* ByteBuddy Helper for a single leaving node failure scenario
*/
- @Shared
public static class BBHelperSingleLeavingNodeFailure
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleTest.java
index 1346688..950fef4 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingSingleTest.java
@@ -36,7 +36,7 @@ import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.analytics.TestUninterruptibles;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.sidecar.testing.QualifiedName;
-import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT;
@@ -95,7 +95,6 @@ class LeavingSingleTest extends LeavingTestBase
/**
* ByteBuddy Helper for a single leaving node
*/
- @Shared
public static class BBHelperSingleLeavingNode
{
static final CountDownLatch transitionalStateStart = new CountDownLatch(1);
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingTestBase.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingTestBase.java
index 7168a9f..2130ef2 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingTestBase.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/shrink/LeavingTestBase.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.analytics.shrink;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -35,13 +34,13 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.analytics.DataGenerationUtils;
import org.apache.cassandra.analytics.ResiliencyTestBase;
import org.apache.cassandra.analytics.TestUninterruptibles;
-import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.testing.utils.ClusterUtils;
+import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.distributed.api.NodeToolResult;
-import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import org.apache.cassandra.spark.bulkwriter.WriterOptions;
-import org.apache.cassandra.testing.TestVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
@@ -74,6 +73,7 @@ abstract class LeavingTestBase extends ResiliencyTestBase
@Override
protected void beforeTestStart()
{
+ super.beforeTestStart();
SparkSession spark = getOrCreateSparkSession();
// Generate some artificial data for the test
df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT);
@@ -82,18 +82,15 @@ abstract class LeavingTestBase extends ResiliencyTestBase
}
@Override
- protected UpgradeableCluster provisionCluster(TestVersion testVersion) throws IOException
+ protected void afterClusterProvisioned()
{
ClusterBuilderConfiguration configuration = testClusterConfiguration();
- UpgradeableCluster provisionedCluster = clusterBuilder(configuration, testVersion);
- IInstance seed = provisionedCluster.getFirstRunningInstance();
- leavingNodes = decommissionNodes(provisionedCluster, leavingNodesPerDc(), configuration.dcCount);
+ IInstance seed = cluster.getFirstRunningInstance();
+ leavingNodes = decommissionNodes(cluster, leavingNodesPerDc(), configuration.dcCount);
// Wait until nodes have reached expected state
TestUninterruptibles.awaitUninterruptiblyOrThrow(transitioningStateStart(), 4, TimeUnit.MINUTES);
- leavingNodes.forEach(instance -> ClusterUtils.awaitRingState(seed, instance, "Leaving"));
-
- return provisionedCluster;
+ leavingNodes.forEach(instance -> cluster.awaitRingState(seed, instance, "Leaving"));
}
protected void completeTransitionsAndValidateWrites(CountDownLatch transitionalStateEnd, Stream<Arguments> testInputs)
@@ -122,11 +119,6 @@ abstract class LeavingTestBase extends ResiliencyTestBase
*/
protected abstract int leavingNodesPerDc();
- /**
- * @return the configuration for the test cluster
- */
- protected abstract ClusterBuilderConfiguration testClusterConfiguration();
-
protected static Stream<Arguments> singleDCTestInputs()
{
return Stream.of(
@@ -146,7 +138,7 @@ abstract class LeavingTestBase extends ResiliencyTestBase
);
}
- protected List<IInstance> decommissionNodes(UpgradeableCluster cluster,
+ protected List<IInstance> decommissionNodes(ICluster<? extends IInstance> cluster,
int leavingNodesPerDC,
int numDcs)
{
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
index 7454180..7f93135 100644
Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index db9a6b8..a80b22c 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,7 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.6-bin.zip
+networkTimeout=10000
+validateDistributionUrl=true
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
index 1b6c787..0adc8e1 100755
--- a/gradlew
+++ b/gradlew
@@ -55,7 +55,7 @@
# Darwin, MinGW, and NonStop.
#
# (3) This script is generated from the Groovy template
-# https://github.com/gradle/gradle/blob/master/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+# https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
# within the Gradle project.
#
# You can find Gradle at https://github.com/gradle/gradle/.
@@ -80,13 +80,11 @@ do
esac
done
-APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
-
-APP_NAME="Gradle"
+# This is normally unused
+# shellcheck disable=SC2034
APP_BASE_NAME=${0##*/}
-
-# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+# Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036)
+APP_HOME=$( cd "${APP_HOME:-./}" > /dev/null && pwd -P ) || exit
# Use the maximum available, or set MAX_FD != -1 to use that value.
MAX_FD=maximum
@@ -133,22 +131,29 @@ location of your Java installation."
fi
else
JAVACMD=java
- which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+ if ! command -v java >/dev/null 2>&1
+ then
+ die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
Please set the JAVA_HOME variable in your environment to match the
location of your Java installation."
+ fi
fi
# Increase the maximum file descriptors if we can.
if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
case $MAX_FD in #(
max*)
+ # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
+ # shellcheck disable=SC3045
MAX_FD=$( ulimit -H -n ) ||
warn "Could not query maximum file descriptor limit"
esac
case $MAX_FD in #(
'' | soft) :;; #(
*)
+ # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
+ # shellcheck disable=SC3045
ulimit -n "$MAX_FD" ||
warn "Could not set maximum file descriptor limit to $MAX_FD"
esac
@@ -193,6 +198,10 @@ if "$cygwin" || "$msys" ; then
done
fi
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+
# Collect all arguments for the java command;
# * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of
# shell script including quotes and variable substitutions, so put them in
@@ -205,6 +214,12 @@ set -- \
org.gradle.wrapper.GradleWrapperMain \
"$@"
+# Stop when "xargs" is not available.
+if ! command -v xargs >/dev/null 2>&1
+then
+ die "xargs is not available"
+fi
+
# Use "xargs" to parse quoted args.
#
# With -n1 it outputs one arg per line, with the quotes and backslashes removed.
diff --git a/scripts/build-sidecar.sh b/scripts/build-sidecar.sh
index 821e7e0..56dc5df 100755
--- a/scripts/build-sidecar.sh
+++ b/scripts/build-sidecar.sh
@@ -24,7 +24,7 @@ else
SCRIPT_DIR=$( dirname -- "$( readlink -f -- "$0"; )"; )
SIDECAR_REPO="${SIDECAR_REPO:-https://github.com/apache/cassandra-sidecar.git}"
SIDECAR_BRANCH="${SIDECAR_BRANCH:-trunk}"
- SIDECAR_COMMIT="${SIDECAR_COMMIT:-056faad8278365a77a36dd336e002136c269165a}"
+ SIDECAR_COMMIT="${SIDECAR_COMMIT:-20795db4d708b9287e0a2281695923bfb6fa9138}"
SIDECAR_JAR_DIR="$(dirname "${SCRIPT_DIR}/")/dependencies"
SIDECAR_JAR_DIR=${CASSANDRA_DEP_DIR:-$SIDECAR_JAR_DIR}
SIDECAR_BUILD_DIR="${SIDECAR_JAR_DIR}/sidecar-build"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org