You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by yc...@apache.org on 2024/09/18 16:33:26 UTC
(cassandra-analytics) branch trunk updated: CASSANDRA-19910: Support data partitioning for multiple clusters coordinated write (#80)
This is an automated email from the ASF dual-hosted git repository.
ycai 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 4fb1e7f CASSANDRA-19910: Support data partitioning for multiple clusters coordinated write (#80)
4fb1e7f is described below
commit 4fb1e7f47d640353cd57f7a3035c70099049b29c
Author: Yifan Cai <yc...@apache.org>
AuthorDate: Wed Sep 18 09:33:21 2024 -0700
CASSANDRA-19910: Support data partitioning for multiple clusters coordinated write (#80)
Patch by Yifan Cai; Reviewed by Doug Rohrer for CASSANDRA-19910
---
.circleci/config.yml | 2 +
CHANGES.txt | 1 +
cassandra-analytics-common/build.gradle | 27 +-
.../cassandra/bridge/CassandraVersionFeatures.java | 21 +-
.../org/apache/cassandra/spark/data/CqlTable.java | 2 +-
.../spark/exception/SidecarApiCallException.java | 2 +
.../spark/exception/TimeSkewTooLargeException.java | 41 +++
cassandra-analytics-core/build.gradle | 2 +-
...Context.java => AbstractBulkWriterContext.java} | 298 +++++++++++----------
.../spark/bulkwriter/BulkWriteValidator.java | 7 +-
.../spark/bulkwriter/BulkWriterContextFactory.java | 94 +++++++
.../bulkwriter/CassandraBulkWriterContext.java | 232 ++--------------
.../spark/bulkwriter/CassandraClusterInfo.java | 88 ++++--
.../spark/bulkwriter/CassandraSchemaInfo.java | 4 +-
.../cassandra/spark/bulkwriter/ClusterInfo.java | 21 +-
.../spark/bulkwriter/DirectStreamSession.java | 2 +-
.../bulkwriter/ImportCompletionCoordinator.java | 2 +-
.../cassandra/spark/bulkwriter/RecordWriter.java | 31 +--
.../cassandra/spark/bulkwriter/RingInstance.java | 2 +-
.../bulkwriter/blobupload/BlobStreamSession.java | 2 +-
.../CassandraClusterInfoGroup.java | 269 +++++++++++++++++++
.../CassandraCoordinatedBulkWriterContext.java | 98 +++++++
.../coordinatedwrite/MultiClusterInfoProvider.java | 28 +-
.../token/ReplicaAwareFailureHandler.java | 12 +-
.../spark/bulkwriter/token/TokenRangeMapping.java | 41 ++-
.../spark/sparksql/CassandraDataSink.java | 16 +-
.../bridge/CassandraVersionFeaturesTest.java | 1 +
.../spark/bulkwriter/BulkWriteValidatorTest.java | 7 +-
.../bulkwriter/BulkWriterContextFactoryTest.java | 101 +++++++
.../spark/bulkwriter/CassandraClusterInfoTest.java | 116 ++++++++
.../spark/bulkwriter/DirectStreamSessionTest.java | 7 +-
.../ImportCompletionCoordinatorTest.java | 7 +-
.../spark/bulkwriter/MockBulkWriterContext.java | 47 ++--
.../spark/bulkwriter/RecordWriterTest.java | 37 ++-
.../spark/bulkwriter/RingInstanceTest.java | 3 +-
.../bulkwriter/StreamSessionConsistencyTest.java | 6 +-
.../spark/bulkwriter/TokenRangeMappingUtils.java | 8 +-
.../CassandraClusterInfoGroupTest.java | 258 ++++++++++++++++++
.../bulkwriter/token/TokenRangeMappingTest.java | 155 +++++++++--
.../build.gradle | 4 +-
.../spark/data/complex/CqlCollection.java | 3 +-
gradle.properties | 1 +
42 files changed, 1557 insertions(+), 549 deletions(-)
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 233c35a..a039e89 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -125,6 +125,7 @@ jobs:
name: Build dependencies for jdk8 builds
command: |
./scripts/build-dependencies.sh
+ JDK_VERSION=1.8 SPARK_VERSION=2 SCALA_VERSION=2.12 ./gradlew codeCheckTasks
- persist_to_workspace:
root: dependencies
paths:
@@ -142,6 +143,7 @@ jobs:
name: Build dependencies for jdk11 builds
command: |
CASSANDRA_USE_JDK11=true ./scripts/build-dependencies.sh
+ ./gradlew codeCheckTasks
- persist_to_workspace:
root: dependencies
paths:
diff --git a/CHANGES.txt b/CHANGES.txt
index 8e70567..9d7288a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.0.0
+ * Support data partitioning for multiple clusters coordinated write (CASSANDRA-19910)
* Add writer option COORDINATED_WRITE_CONFIG to define coordinated write to multiple Cassandra clusters (CASSANDRA-19909)
* Decouple Cassandra types from Spark types so Cassandra types can be used independently from Spark (CASSANDRA-19815)
* Make the compression cache configurable to reduce heap pressure for large SSTables (CASSANDRA-19900)
diff --git a/cassandra-analytics-common/build.gradle b/cassandra-analytics-common/build.gradle
index 1fc95bc..c92b1bd 100644
--- a/cassandra-analytics-common/build.gradle
+++ b/cassandra-analytics-common/build.gradle
@@ -41,20 +41,21 @@ publishing {
}
dependencies {
- implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
- compileOnly "com.esotericsoftware:kryo-shaded:${kryoVersion}"
- compileOnly "com.google.guava:guava:${guavaVersion}"
- compileOnly "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
- compileOnly "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}"
- compileOnly "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
+ implementation("org.slf4j:slf4j-api:${slf4jApiVersion}")
+ compileOnly("com.esotericsoftware:kryo-shaded:${kryoVersion}")
+ compileOnly("com.google.guava:guava:${guavaVersion}")
+ compileOnly("com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}")
+ compileOnly("com.fasterxml.jackson.core:jackson-core:${jacksonVersion}")
+ compileOnly("com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}")
- testImplementation "com.google.guava:guava:${guavaVersion}"
- testImplementation "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
- testImplementation "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}"
- testImplementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
- 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("com.google.guava:guava:${guavaVersion}")
+ testImplementation("com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}")
+ testImplementation("com.fasterxml.jackson.core:jackson-core:${jacksonVersion}")
+ testImplementation("com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}")
+ testImplementation("org.junit.jupiter:junit-jupiter-api:${junitVersion}")
+ testImplementation("org.junit.jupiter:junit-jupiter-params:${junitVersion}")
+ testImplementation("org.junit.jupiter:junit-jupiter-engine:${junitVersion}")
+ testImplementation("org.assertj:assertj-core:${assertjCoreVersion}")
testImplementation("commons-io:commons-io:2.8.0")
}
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersionFeatures.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersionFeatures.java
index 089f574..d2d59ea 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersionFeatures.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersionFeatures.java
@@ -23,6 +23,8 @@ import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import com.google.common.annotations.VisibleForTesting;
+
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -36,11 +38,20 @@ public class CassandraVersionFeatures implements Comparable<CassandraVersionFeat
protected final int minorVersion;
protected final String suffix;
+ private final String rawVersionString;
+
+ @VisibleForTesting
public CassandraVersionFeatures(int majorVersion, int minorVersion, @Nullable String suffix)
+ {
+ this(majorVersion, minorVersion, suffix, null);
+ }
+
+ public CassandraVersionFeatures(int majorVersion, int minorVersion, @Nullable String suffix, @Nullable String rawVersionString)
{
this.majorVersion = majorVersion;
this.minorVersion = minorVersion;
this.suffix = suffix;
+ this.rawVersionString = rawVersionString;
}
/**
@@ -49,14 +60,13 @@ public class CassandraVersionFeatures implements Comparable<CassandraVersionFeat
* @param cassandraVersion the string representing the cassandra version
* @return an instance representing the parsed values from the version string
*/
- public static CassandraVersionFeatures cassandraVersionFeaturesFromCassandraVersion(
- @NotNull String cassandraVersion)
+ public static CassandraVersionFeatures cassandraVersionFeaturesFromCassandraVersion(@NotNull String cassandraVersion)
{
String versionCode = getCassandraVersionCode(cassandraVersion);
String minorVersionCode = getCassandraMinorVersionCode(cassandraVersion);
String versionSuffix = getCassandraVersionSuffix(cassandraVersion);
- return new CassandraVersionFeatures(Integer.parseInt(versionCode), Integer.parseInt(minorVersionCode), versionSuffix);
+ return new CassandraVersionFeatures(Integer.parseInt(versionCode), Integer.parseInt(minorVersionCode), versionSuffix, cassandraVersion);
}
// E.g if cassandra version = cassandra-1.2.11-v1, we return 12;
@@ -139,6 +149,11 @@ public class CassandraVersionFeatures implements Comparable<CassandraVersionFeat
return suffix;
}
+ public String getRawVersionString()
+ {
+ return rawVersionString;
+ }
+
@Override
public int compareTo(@NotNull CassandraVersionFeatures that)
{
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java
index 1cd4f25..a0bc79b 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java
@@ -39,7 +39,7 @@ import org.jetbrains.annotations.NotNull;
@SuppressWarnings({ "WeakerAccess", "unused" })
public class CqlTable implements Serializable
{
- public static final long serialVersionUID = 42L;
+ private static final long serialVersionUID = 1018995207366817661L;
private final ReplicationFactor replicationFactor;
private final String keyspace;
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/SidecarApiCallException.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/SidecarApiCallException.java
index d59c138..14ea7b7 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/SidecarApiCallException.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/SidecarApiCallException.java
@@ -24,6 +24,8 @@ package org.apache.cassandra.spark.exception;
*/
public class SidecarApiCallException extends AnalyticsException
{
+ private static final long serialVersionUID = 3304206898661966469L;
+
public SidecarApiCallException(String message, Throwable cause)
{
super(message, cause);
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/TimeSkewTooLargeException.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/TimeSkewTooLargeException.java
new file mode 100644
index 0000000..9e7a54e
--- /dev/null
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/TimeSkewTooLargeException.java
@@ -0,0 +1,41 @@
+/*
+ * 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.spark.exception;
+
+import java.time.Instant;
+
+/**
+ * Exception for the case when the time drift between local (spark) and remote (cassandra) is too large
+ */
+public class TimeSkewTooLargeException extends AnalyticsException
+{
+ private static final long serialVersionUID = -6748770894292325624L;
+
+ public TimeSkewTooLargeException(int allowableDurationMinutes, Instant localNow, Instant remoteNow, String clusterId)
+ {
+ super(makeExceptionMessage("Time skew between Spark and Cassandra is too large", allowableDurationMinutes, localNow, remoteNow, clusterId));
+ }
+
+ private static String makeExceptionMessage(String summary, int allowableDurationMinutes, Instant localNow, Instant remoteNow, String clusterId)
+ {
+ return String.format("%s. allowableSkewInMinutes=%d, localTime=%s, remoteCassandraTime=%s, clusterId=%s",
+ summary, allowableDurationMinutes, localNow, remoteNow, clusterId);
+ }
+}
diff --git a/cassandra-analytics-core/build.gradle b/cassandra-analytics-core/build.gradle
index c74854d..b0f9c9c 100644
--- a/cassandra-analytics-core/build.gradle
+++ b/cassandra-analytics-core/build.gradle
@@ -94,7 +94,7 @@ dependencies {
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("org.assertj:assertj-core:${assertjCoreVersion}")
testImplementation(group: 'org.quicktheories', name: 'quicktheories', version: "${project.rootProject.quickTheoriesVersion}")
testImplementation(group: 'org.slf4j', name: 'slf4j-simple', version: '1.7.26')
testImplementation(group: 'org.mockito', name: 'mockito-core', version: "${project.rootProject.mockitoVersion}")
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/AbstractBulkWriterContext.java
similarity index 51%
copy from cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java
copy to cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/AbstractBulkWriterContext.java
index 632b6c0..4b9f9f0 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/AbstractBulkWriterContext.java
@@ -19,11 +19,10 @@
package org.apache.cassandra.spark.bulkwriter;
-import java.util.HashMap;
-import java.util.Map;
import java.util.Set;
+import java.util.UUID;
+import java.util.function.Supplier;
-import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,61 +36,108 @@ import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.common.stats.JobStatsPublisher;
import org.apache.cassandra.spark.common.stats.LogStatsPublisher;
import org.apache.cassandra.spark.data.CqlTable;
+import org.apache.cassandra.spark.data.QualifiedTableName;
import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
import org.apache.cassandra.spark.utils.CqlUtils;
-import org.apache.cassandra.spark.utils.ScalaFunctions;
-import org.apache.spark.SparkContext;
import org.apache.spark.sql.types.StructType;
-import org.apache.spark.util.ShutdownHookManager;
import org.jetbrains.annotations.NotNull;
-// CHECKSTYLE IGNORE: This class cannot be declared as final, because consumers should be able to extend it
-public class CassandraBulkWriterContext implements BulkWriterContext, KryoSerializable
+public abstract class AbstractBulkWriterContext implements BulkWriterContext, KryoSerializable
{
- private static final long serialVersionUID = 8241993502687688783L;
- private static final Logger LOGGER = LoggerFactory.getLogger(CassandraBulkWriterContext.class);
- @NotNull
+ private static final long serialVersionUID = -6526396615116954510L;
+ // log as the concrete implementation; but use private to not expose the logger to implementations
+ private final transient Logger logger = LoggerFactory.getLogger(this.getClass());
+
private final BulkSparkConf conf;
+ private final int sparkDefaultParallelism;
private final JobInfo jobInfo;
- private final String lowestCassandraVersion;
- private transient CassandraBridge bridge;
- private final CassandraClusterInfo clusterInfo;
+ private final ClusterInfo clusterInfo;
private final SchemaInfo schemaInfo;
- private final transient JobStatsPublisher jobStatsPublisher;
- protected transient volatile TransportContext transportContext;
+ private final String lowestCassandraVersion;
+ // Note: do not declare transient fields as final; but they need to be volatile as there could be contention when recreating them after deserialization
+ // For the transient field, they are assigned null once deserialized, remember to use getOrRebuildAfterDeserialization for their getters
+ private transient volatile CassandraBridge bridge;
+ private transient volatile JobStatsPublisher jobStatsPublisher;
+ private transient volatile TransportContext transportContext;
- protected CassandraBulkWriterContext(@NotNull BulkSparkConf conf,
- @NotNull CassandraClusterInfo clusterInfo,
- @NotNull StructType dfSchema,
- SparkContext sparkContext)
+ protected AbstractBulkWriterContext(@NotNull BulkSparkConf conf,
+ @NotNull StructType structType,
+ @NotNull int sparkDefaultParallelism)
{
this.conf = conf;
- this.clusterInfo = clusterInfo;
- clusterInfo.startupValidate();
- this.jobStatsPublisher = new LogStatsPublisher();
- lowestCassandraVersion = clusterInfo.getLowestCassandraVersion();
- this.bridge = CassandraBridgeFactory.get(lowestCassandraVersion);
- TokenRangeMapping<RingInstance> tokenRangeMapping = clusterInfo.getTokenRangeMapping(true);
- jobInfo = new CassandraJobInfo(conf,
- bridge.getTimeUUID(), // used for creating restore job on sidecar
- new TokenPartitioner(tokenRangeMapping,
- conf.numberSplits,
- sparkContext.defaultParallelism(),
- conf.getCores()));
- Preconditions.checkArgument(!conf.consistencyLevel.isLocal()
- || (conf.localDC != null && tokenRangeMapping.replicationFactor()
- .getOptions()
- .containsKey(conf.localDC)),
- String.format("Keyspace %s is not replicated on datacenter %s", conf.keyspace, conf.localDC));
-
- transportContext = createTransportContext(true);
-
- String keyspace = jobInfo.qualifiedTableName().keyspace();
- String table = jobInfo.qualifiedTableName().table();
-
- String keyspaceSchema = clusterInfo.getKeyspaceSchema(true);
- Partitioner partitioner = clusterInfo.getPartitioner();
+ this.sparkDefaultParallelism = sparkDefaultParallelism;
+ // Note: build sequence matters
+ this.clusterInfo = buildClusterInfo();
+ this.lowestCassandraVersion = findLowestCassandraVersion();
+ this.bridge = buildCassandraBridge();
+ this.jobInfo = buildJobInfo();
+ this.schemaInfo = buildSchemaInfo(structType);
+ this.jobStatsPublisher = buildJobStatsPublisher();
+ this.transportContext = buildTransportContext(true);
+ }
+
+ protected final BulkSparkConf bulkSparkConf()
+ {
+ return conf;
+ }
+
+ protected final int sparkDefaultParallelism()
+ {
+ return sparkDefaultParallelism;
+ }
+
+ protected String lowestCassandraVersion()
+ {
+ return lowestCassandraVersion;
+ }
+
+ /*--- Methods to build required fields ---*/
+
+ protected abstract ClusterInfo buildClusterInfo();
+
+ protected abstract void validateKeyspaceReplication();
+
+ protected JobInfo buildJobInfo()
+ {
+ validateKeyspaceReplication();
+ BulkSparkConf conf = bulkSparkConf();
+ TokenRangeMapping<RingInstance> tokenRangeMapping = cluster().getTokenRangeMapping(true);
+ UUID restoreJobId = bridge().getTimeUUID(); // used for creating restore job on sidecar
+ TokenPartitioner tokenPartitioner = new TokenPartitioner(tokenRangeMapping,
+ conf.numberSplits,
+ sparkDefaultParallelism(),
+ conf.getCores());
+ return new CassandraJobInfo(conf, restoreJobId, tokenPartitioner);
+ }
+
+ protected CassandraBridge buildCassandraBridge()
+ {
+ return CassandraBridgeFactory.get(lowestCassandraVersion());
+ }
+
+ protected TransportContext buildTransportContext(boolean isOnDriver)
+ {
+ return createTransportContext(isOnDriver);
+ }
+
+ protected JobStatsPublisher buildJobStatsPublisher()
+ {
+ return new LogStatsPublisher();
+ }
+
+ protected String findLowestCassandraVersion()
+ {
+ return cluster().getLowestCassandraVersion();
+ }
+
+ protected SchemaInfo buildSchemaInfo(StructType structType)
+ {
+ QualifiedTableName tableName = job().qualifiedTableName();
+ String keyspace = tableName.keyspace();
+ String table = tableName.table();
+ String keyspaceSchema = cluster().getKeyspaceSchema(true);
+ Partitioner partitioner = cluster().getPartitioner();
String createTableSchema = CqlUtils.extractTableSchema(keyspaceSchema, keyspace, table);
Set<String> udts = CqlUtils.extractUdts(keyspaceSchema, keyspace);
ReplicationFactor replicationFactor = CqlUtils.extractReplicationFactor(keyspaceSchema, keyspace);
@@ -99,54 +145,56 @@ public class CassandraBulkWriterContext implements BulkWriterContext, KryoSerial
CqlTable cqlTable = bridge().buildSchema(createTableSchema, keyspace, replicationFactor, partitioner, udts, null, indexCount);
TableInfoProvider tableInfoProvider = new CqlTableInfoProvider(createTableSchema, cqlTable);
- TableSchema tableSchema = initializeTableSchema(conf, dfSchema, tableInfoProvider, lowestCassandraVersion);
- schemaInfo = new CassandraSchemaInfo(tableSchema, udts, cqlTable);
+ TableSchema tableSchema = initializeTableSchema(bulkSparkConf(), structType, tableInfoProvider, lowestCassandraVersion());
+ return new CassandraSchemaInfo(tableSchema, udts);
+ }
+
+ /*-------------------------------------------*/
+
+ @Override
+ public JobInfo job()
+ {
+ return jobInfo;
+ }
+
+ @Override
+ public ClusterInfo cluster()
+ {
+ return clusterInfo;
+ }
+
+ @Override
+ public SchemaInfo schema()
+ {
+ return schemaInfo;
}
@Override
public CassandraBridge bridge()
{
- CassandraBridge currentBridge = this.bridge;
- if (currentBridge != null)
- {
- return currentBridge;
- }
- this.bridge = CassandraBridgeFactory.get(lowestCassandraVersion);
+ bridge = getOrRebuildAfterDeserialization(() -> bridge, this::buildCassandraBridge);
return bridge;
}
- // Static factory to create BulkWriterContext based on the requested Bulk Writer transport strategy
- public static BulkWriterContext fromOptions(@NotNull SparkContext sparkContext,
- @NotNull Map<String, String> strOptions,
- @NotNull StructType dfSchema)
+ @Override
+ public JobStatsPublisher jobStats()
{
- Preconditions.checkNotNull(dfSchema);
-
- BulkSparkConf conf = new BulkSparkConf(sparkContext.getConf(), strOptions);
- CassandraClusterInfo clusterInfo = new CassandraClusterInfo(conf);
- CassandraBulkWriterContext bulkWriterContext = new CassandraBulkWriterContext(conf, clusterInfo, dfSchema, sparkContext);
- ShutdownHookManager.addShutdownHook(org.apache.spark.util.ShutdownHookManager.TEMP_DIR_SHUTDOWN_PRIORITY(),
- ScalaFunctions.wrapLambda(bulkWriterContext::shutdown));
- bulkWriterContext.publishInitialJobStats(sparkContext.version());
- return bulkWriterContext;
+ jobStatsPublisher = getOrRebuildAfterDeserialization(() -> jobStatsPublisher, this::buildJobStatsPublisher);
+ return jobStatsPublisher;
}
- private void publishInitialJobStats(String sparkVersion)
+ @Override
+ public TransportContext transportContext()
{
- Map<String, String> initialJobStats = new HashMap<String, String>() // type declaration required to compile with java8
- {{
- put("jobId", jobInfo.getId().toString());
- put("sparkVersion", sparkVersion);
- put("keyspace", jobInfo.getId().toString());
- put("table", jobInfo.getId().toString());
- }};
- jobStatsPublisher.publish(initialJobStats);
+ transportContext = getOrRebuildAfterDeserialization(() -> transportContext, () -> buildTransportContext(false));
+ return transportContext;
}
@Override
public void shutdown()
{
- LOGGER.info("Shutting down {}", this);
+ logger.info("Shutting down bulk writer context. contextClass={}", getClass().getSimpleName());
+
if (clusterInfo != null)
{
clusterInfo.close();
@@ -158,9 +206,33 @@ public class CassandraBulkWriterContext implements BulkWriterContext, KryoSerial
}
}
+ @NotNull
+ protected TableSchema initializeTableSchema(@NotNull BulkSparkConf conf,
+ @NotNull StructType dfSchema,
+ TableInfoProvider tableInfoProvider,
+ String lowestCassandraVersion)
+ {
+ return new TableSchema(dfSchema,
+ tableInfoProvider,
+ conf.writeMode,
+ conf.getTTLOptions(),
+ conf.getTimestampOptions(),
+ lowestCassandraVersion,
+ job().qualifiedTableName().quoteIdentifiers());
+ }
+
+ @NotNull
+ protected TransportContext createTransportContext(boolean isOnDriver)
+ {
+ BulkSparkConf conf = bulkSparkConf();
+ return conf.getTransportInfo()
+ .getTransport()
+ .createContext(this, conf, isOnDriver);
+ }
+
/**
- * Use the implementation of the KryoSerializable interface as a detection device to make sure the Spark Bulk
- * Writer's KryoRegistrator is properly in place.
+ * Use the implementation of the KryoSerializable interface as a detection device to make sure
+ * {@link org.apache.cassandra.spark.bulkwriter.util.SbwKryoRegistrator} is properly in place.
* <p>
* If this class is serialized by Kryo, it means we're <b>not</b> set up correctly, and therefore we log and fail.
* This failure will occur early in the job and be very clear, so users can quickly fix their code and get up and
@@ -184,76 +256,28 @@ public class CassandraBulkWriterContext implements BulkWriterContext, KryoSerial
private void failIfKryoNotRegistered()
{
- LOGGER.error(KRYO_REGISTRATION_WARNING);
+ logger.error(KRYO_REGISTRATION_WARNING);
throw new RuntimeException(KRYO_REGISTRATION_WARNING);
}
- @Override
- @NotNull
- public ClusterInfo cluster()
- {
- return clusterInfo;
- }
-
- @Override
- @NotNull
- public JobInfo job()
- {
- return jobInfo;
- }
-
- @NotNull
- public JobStatsPublisher jobStats()
- {
- return jobStatsPublisher;
- }
-
- @Override
- public SchemaInfo schema()
- {
- return schemaInfo;
- }
-
- @Override
- public TransportContext transportContext()
+ // returns immediately if current supplies non-null value; otherwise, it invokes the builder in a synchronized block to only build once
+ private <T> T getOrRebuildAfterDeserialization(Supplier<T> current, Supplier<T> builder)
{
- // When running on driver, transportContext is created at the constructor, and it is not null
- if (transportContext != null)
+ T t = current.get();
+ if (t != null)
{
- return transportContext;
+ return t;
}
- // When running on executor, transportContext is null. Synchronize to avoid multi-instantiation
synchronized (this)
{
- if (transportContext == null)
+ t = current.get();
+ if (t != null)
{
- transportContext = createTransportContext(false);
+ return t;
}
- }
- return transportContext;
- }
-
- @NotNull
- protected TransportContext createTransportContext(boolean isOnDriver)
- {
- return conf.getTransportInfo()
- .getTransport()
- .createContext(this, conf, isOnDriver);
- }
- @NotNull
- protected TableSchema initializeTableSchema(@NotNull BulkSparkConf conf,
- @NotNull StructType dfSchema,
- TableInfoProvider tableInfoProvider,
- String lowestCassandraVersion)
- {
- return new TableSchema(dfSchema,
- tableInfoProvider,
- conf.writeMode,
- conf.getTTLOptions(),
- conf.getTimestampOptions(),
- lowestCassandraVersion,
- conf.quoteIdentifiers);
+ return builder.get();
+ }
}
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java
index accfcc8..56e1989 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java
@@ -54,10 +54,11 @@ public class BulkWriteValidator
ReplicaAwareFailureHandler<RingInstance> failureHandler,
Logger logger,
String phase,
- JobInfo job)
+ JobInfo job,
+ ClusterInfo cluster)
{
List<ReplicaAwareFailureHandler<RingInstance>.ConsistencyFailurePerRange> failedRanges =
- failureHandler.getFailedRanges(tokenRangeMapping, job.getConsistencyLevel(), job.getLocalDC());
+ failureHandler.getFailedRanges(tokenRangeMapping, job.getConsistencyLevel(), job.getLocalDC(), cluster.replicationFactor());
if (failedRanges.isEmpty())
{
@@ -142,7 +143,7 @@ public class BulkWriteValidator
updateInstanceAvailability();
}
// Fails if the failures violate consistency requirements
- validateClOrFail(tokenRangeMapping, failureHandler, LOGGER, phase, job);
+ validateClOrFail(tokenRangeMapping, failureHandler, LOGGER, phase, job, cluster);
}
private void updateInstanceAvailability()
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriterContextFactory.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriterContextFactory.java
new file mode 100644
index 0000000..9d710bd
--- /dev/null
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriterContextFactory.java
@@ -0,0 +1,94 @@
+/*
+ * 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.spark.bulkwriter;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.common.base.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.spark.bulkwriter.coordinatedwrite.CassandraCoordinatedBulkWriterContext;
+import org.apache.cassandra.spark.utils.ScalaFunctions;
+import org.apache.spark.SparkContext;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.util.ShutdownHookManager;
+import org.jetbrains.annotations.NotNull;
+
+public class BulkWriterContextFactory
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(BulkWriterContextFactory.class);
+
+ public static final BulkWriterContextFactory INSTANCE = new BulkWriterContextFactory();
+
+ @NotNull
+ public BulkWriterContext createBulkWriterContext(@NotNull SparkContext sparkContext,
+ @NotNull Map<String, String> options,
+ @NotNull StructType schema)
+ {
+ Preconditions.checkNotNull(schema);
+
+ BulkWriterContext bulkWriterContext;
+ BulkSparkConf conf = new BulkSparkConf(sparkContext.getConf(), options);
+ int sparkDefaultParallelism = sparkContext.defaultParallelism();
+ if (conf.isCoordinatedWriteConfigured())
+ {
+ LOGGER.info("Initializing bulk writer context for multi-clusters coordinated write");
+ bulkWriterContext = createCoordinatedBulkWriterContext(conf, schema, sparkDefaultParallelism);
+ }
+ else
+ {
+ LOGGER.info("Initializing bulk writer context for single cluster write");
+ bulkWriterContext = createBulkWriterContext(conf, schema, sparkDefaultParallelism);
+ }
+
+ ShutdownHookManager.addShutdownHook(org.apache.spark.util.ShutdownHookManager.TEMP_DIR_SHUTDOWN_PRIORITY(),
+ ScalaFunctions.wrapLambda(bulkWriterContext::shutdown));
+ publishInitialJobStats(bulkWriterContext, sparkContext.version());
+ return bulkWriterContext;
+ }
+
+ @NotNull
+ protected BulkWriterContext createBulkWriterContext(@NotNull BulkSparkConf conf, StructType schema, int sparkDefaultParallelism)
+ {
+ return new CassandraBulkWriterContext(conf, schema, sparkDefaultParallelism);
+ }
+
+ @NotNull
+ protected BulkWriterContext createCoordinatedBulkWriterContext(@NotNull BulkSparkConf conf, StructType schema, int sparkDefaultParallelism)
+ {
+ return new CassandraCoordinatedBulkWriterContext(conf, schema, sparkDefaultParallelism);
+ }
+
+ protected void publishInitialJobStats(BulkWriterContext context, String sparkVersion)
+ {
+ JobInfo jobInfo = context.job();
+ Map<String, String> initialJobStats = new HashMap<String, String>() // type declaration required to compile with java8
+ {{
+ put("jobId", jobInfo.getId());
+ put("sparkVersion", sparkVersion);
+ put("keyspace", jobInfo.qualifiedTableName().keyspace());
+ put("table", jobInfo.qualifiedTableName().table());
+ }};
+ context.jobStats().publish(initialJobStats);
+ }
+}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java
index 632b6c0..bd61a29 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java
@@ -19,241 +19,45 @@
package org.apache.cassandra.spark.bulkwriter;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
import com.google.common.base.Preconditions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.commons.lang3.StringUtils;
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.KryoSerializable;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
-import org.apache.cassandra.bridge.CassandraBridge;
-import org.apache.cassandra.bridge.CassandraBridgeFactory;
-import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
-import org.apache.cassandra.spark.common.stats.JobStatsPublisher;
-import org.apache.cassandra.spark.common.stats.LogStatsPublisher;
-import org.apache.cassandra.spark.data.CqlTable;
-import org.apache.cassandra.spark.data.ReplicationFactor;
-import org.apache.cassandra.spark.data.partitioner.Partitioner;
-import org.apache.cassandra.spark.utils.CqlUtils;
-import org.apache.cassandra.spark.utils.ScalaFunctions;
-import org.apache.spark.SparkContext;
import org.apache.spark.sql.types.StructType;
-import org.apache.spark.util.ShutdownHookManager;
import org.jetbrains.annotations.NotNull;
// CHECKSTYLE IGNORE: This class cannot be declared as final, because consumers should be able to extend it
-public class CassandraBulkWriterContext implements BulkWriterContext, KryoSerializable
+public class CassandraBulkWriterContext extends AbstractBulkWriterContext
{
private static final long serialVersionUID = 8241993502687688783L;
- private static final Logger LOGGER = LoggerFactory.getLogger(CassandraBulkWriterContext.class);
- @NotNull
- private final BulkSparkConf conf;
- private final JobInfo jobInfo;
- private final String lowestCassandraVersion;
- private transient CassandraBridge bridge;
- private final CassandraClusterInfo clusterInfo;
- private final SchemaInfo schemaInfo;
- private final transient JobStatsPublisher jobStatsPublisher;
- protected transient volatile TransportContext transportContext;
protected CassandraBulkWriterContext(@NotNull BulkSparkConf conf,
- @NotNull CassandraClusterInfo clusterInfo,
- @NotNull StructType dfSchema,
- SparkContext sparkContext)
+ @NotNull StructType structType,
+ int sparkDefaultParallelism)
{
- this.conf = conf;
- this.clusterInfo = clusterInfo;
- clusterInfo.startupValidate();
- this.jobStatsPublisher = new LogStatsPublisher();
- lowestCassandraVersion = clusterInfo.getLowestCassandraVersion();
- this.bridge = CassandraBridgeFactory.get(lowestCassandraVersion);
- TokenRangeMapping<RingInstance> tokenRangeMapping = clusterInfo.getTokenRangeMapping(true);
- jobInfo = new CassandraJobInfo(conf,
- bridge.getTimeUUID(), // used for creating restore job on sidecar
- new TokenPartitioner(tokenRangeMapping,
- conf.numberSplits,
- sparkContext.defaultParallelism(),
- conf.getCores()));
- Preconditions.checkArgument(!conf.consistencyLevel.isLocal()
- || (conf.localDC != null && tokenRangeMapping.replicationFactor()
- .getOptions()
- .containsKey(conf.localDC)),
- String.format("Keyspace %s is not replicated on datacenter %s", conf.keyspace, conf.localDC));
-
- transportContext = createTransportContext(true);
-
- String keyspace = jobInfo.qualifiedTableName().keyspace();
- String table = jobInfo.qualifiedTableName().table();
-
- String keyspaceSchema = clusterInfo.getKeyspaceSchema(true);
- Partitioner partitioner = clusterInfo.getPartitioner();
- String createTableSchema = CqlUtils.extractTableSchema(keyspaceSchema, keyspace, table);
- Set<String> udts = CqlUtils.extractUdts(keyspaceSchema, keyspace);
- ReplicationFactor replicationFactor = CqlUtils.extractReplicationFactor(keyspaceSchema, keyspace);
- int indexCount = CqlUtils.extractIndexCount(keyspaceSchema, keyspace, table);
- CqlTable cqlTable = bridge().buildSchema(createTableSchema, keyspace, replicationFactor, partitioner, udts, null, indexCount);
-
- TableInfoProvider tableInfoProvider = new CqlTableInfoProvider(createTableSchema, cqlTable);
- TableSchema tableSchema = initializeTableSchema(conf, dfSchema, tableInfoProvider, lowestCassandraVersion);
- schemaInfo = new CassandraSchemaInfo(tableSchema, udts, cqlTable);
+ super(conf, structType, sparkDefaultParallelism);
}
@Override
- public CassandraBridge bridge()
- {
- CassandraBridge currentBridge = this.bridge;
- if (currentBridge != null)
- {
- return currentBridge;
- }
- this.bridge = CassandraBridgeFactory.get(lowestCassandraVersion);
- return bridge;
- }
-
- // Static factory to create BulkWriterContext based on the requested Bulk Writer transport strategy
- public static BulkWriterContext fromOptions(@NotNull SparkContext sparkContext,
- @NotNull Map<String, String> strOptions,
- @NotNull StructType dfSchema)
- {
- Preconditions.checkNotNull(dfSchema);
-
- BulkSparkConf conf = new BulkSparkConf(sparkContext.getConf(), strOptions);
- CassandraClusterInfo clusterInfo = new CassandraClusterInfo(conf);
- CassandraBulkWriterContext bulkWriterContext = new CassandraBulkWriterContext(conf, clusterInfo, dfSchema, sparkContext);
- ShutdownHookManager.addShutdownHook(org.apache.spark.util.ShutdownHookManager.TEMP_DIR_SHUTDOWN_PRIORITY(),
- ScalaFunctions.wrapLambda(bulkWriterContext::shutdown));
- bulkWriterContext.publishInitialJobStats(sparkContext.version());
- return bulkWriterContext;
- }
-
- private void publishInitialJobStats(String sparkVersion)
- {
- Map<String, String> initialJobStats = new HashMap<String, String>() // type declaration required to compile with java8
- {{
- put("jobId", jobInfo.getId().toString());
- put("sparkVersion", sparkVersion);
- put("keyspace", jobInfo.getId().toString());
- put("table", jobInfo.getId().toString());
- }};
- jobStatsPublisher.publish(initialJobStats);
- }
-
- @Override
- public void shutdown()
- {
- LOGGER.info("Shutting down {}", this);
- if (clusterInfo != null)
- {
- clusterInfo.close();
- }
-
- if (transportContext != null)
- {
- transportContext.close();
- }
- }
-
- /**
- * Use the implementation of the KryoSerializable interface as a detection device to make sure the Spark Bulk
- * Writer's KryoRegistrator is properly in place.
- * <p>
- * If this class is serialized by Kryo, it means we're <b>not</b> set up correctly, and therefore we log and fail.
- * This failure will occur early in the job and be very clear, so users can quickly fix their code and get up and
- * running again, rather than having a random NullPointerException further down the line.
- */
- public static final String KRYO_REGISTRATION_WARNING =
- "Spark Bulk Writer Kryo Registrator (SbwKryoRegistrator) was not registered with Spark - "
- + "please see the README.md file for more details on how to register the Spark Bulk Writer.";
-
- @Override
- public void write(Kryo kryo, Output output)
- {
- failIfKryoNotRegistered();
- }
-
- @Override
- public void read(Kryo kryo, Input input)
- {
- failIfKryoNotRegistered();
- }
-
- private void failIfKryoNotRegistered()
- {
- LOGGER.error(KRYO_REGISTRATION_WARNING);
- throw new RuntimeException(KRYO_REGISTRATION_WARNING);
- }
-
- @Override
- @NotNull
- public ClusterInfo cluster()
+ protected ClusterInfo buildClusterInfo()
{
+ ClusterInfo clusterInfo = new CassandraClusterInfo(bulkSparkConf());
+ // perform validation at the earliest. The subsequent statements can invoke network calls
+ clusterInfo.startupValidate();
return clusterInfo;
}
@Override
- @NotNull
- public JobInfo job()
- {
- return jobInfo;
- }
-
- @NotNull
- public JobStatsPublisher jobStats()
- {
- return jobStatsPublisher;
- }
-
- @Override
- public SchemaInfo schema()
- {
- return schemaInfo;
- }
-
- @Override
- public TransportContext transportContext()
+ protected void validateKeyspaceReplication()
{
- // When running on driver, transportContext is created at the constructor, and it is not null
- if (transportContext != null)
+ BulkSparkConf conf = bulkSparkConf();
+ // no validation for non-local CL
+ if (!conf.consistencyLevel.isLocal())
{
- return transportContext;
+ return;
}
-
- // When running on executor, transportContext is null. Synchronize to avoid multi-instantiation
- synchronized (this)
- {
- if (transportContext == null)
- {
- transportContext = createTransportContext(false);
- }
- }
- return transportContext;
- }
-
- @NotNull
- protected TransportContext createTransportContext(boolean isOnDriver)
- {
- return conf.getTransportInfo()
- .getTransport()
- .createContext(this, conf, isOnDriver);
- }
-
- @NotNull
- protected TableSchema initializeTableSchema(@NotNull BulkSparkConf conf,
- @NotNull StructType dfSchema,
- TableInfoProvider tableInfoProvider,
- String lowestCassandraVersion)
- {
- return new TableSchema(dfSchema,
- tableInfoProvider,
- conf.writeMode,
- conf.getTTLOptions(),
- conf.getTimestampOptions(),
- lowestCassandraVersion,
- conf.quoteIdentifiers);
+ // localDc is not empty and replication option contains localDc
+ boolean isReplicatedToLocalDc = !StringUtils.isEmpty(conf.localDC)
+ && cluster().replicationFactor().getOptions().containsKey(conf.localDC);
+ Preconditions.checkState(isReplicatedToLocalDc, "Keyspace %s is not replicated on datacenter %s", conf.keyspace, conf.localDC);
}
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
index 4165c7c..5954671 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
@@ -20,7 +20,11 @@
package org.apache.cassandra.spark.bulkwriter;
import java.io.Closeable;
+import java.math.BigInteger;
+import java.time.Duration;
+import java.time.Instant;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -33,6 +37,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Range;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,9 +55,9 @@ import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
import org.apache.cassandra.spark.exception.SidecarApiCallException;
+import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
import org.apache.cassandra.spark.utils.CqlUtils;
import org.apache.cassandra.spark.utils.FutureUtils;
-import org.jetbrains.annotations.NotNull;
import static org.apache.cassandra.bridge.CassandraBridgeFactory.maybeQuotedIdentifier;
@@ -68,6 +73,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
protected transient volatile TokenRangeMapping<RingInstance> tokenRangeReplicas;
protected transient volatile String keyspaceSchema;
+ protected transient volatile ReplicationFactor replicationFactor;
protected transient volatile CassandraContext cassandraContext;
protected final transient AtomicReference<NodeSettings> nodeSettings;
protected final transient List<CompletableFuture<NodeSettings>> allNodeSettingFutures;
@@ -86,7 +92,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
LOGGER.info("Getting Cassandra versions from all nodes");
this.nodeSettings = new AtomicReference<>(null);
this.allNodeSettingFutures = Sidecar.allNodeSettings(cassandraContext.getSidecarClient(),
- cassandraContext.clusterConfig);
+ cassandraContext.getCluster());
}
@Override
@@ -186,19 +192,38 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
}
@Override
- public TimeSkewResponse getTimeSkew(List<RingInstance> replicas)
+ public void validateTimeSkew(Range<BigInteger> range) throws SidecarApiCallException, TimeSkewTooLargeException
{
+ validateTimeSkewWithLocalNow(range, Instant.now());
+ }
+
+ @VisibleForTesting
+ void validateTimeSkewWithLocalNow(Range<BigInteger> range, Instant localNow) throws SidecarApiCallException, TimeSkewTooLargeException
+ {
+ TimeSkewResponse timeSkew;
try
{
- List<SidecarInstance> instances = replicas
- .stream()
- .map(replica -> new SidecarInstanceImpl(replica.nodeName(), getCassandraContext().sidecarPort()))
- .collect(Collectors.toList());
- return getCassandraContext().getSidecarClient().timeSkew(instances).get();
+ TokenRangeMapping<RingInstance> topology = getTokenRangeMapping(true);
+ List<SidecarInstance> instances = topology.getSubRanges(range)
+ .asMapOfRanges()
+ .values()
+ .stream()
+ .flatMap(Collection::stream)
+ .distinct() // remove duplications
+ .map(replica -> new SidecarInstanceImpl(replica.nodeName(), getCassandraContext().sidecarPort()))
+ .collect(Collectors.toList());
+ timeSkew = getCassandraContext().getSidecarClient().timeSkew(instances).get();
}
catch (InterruptedException | ExecutionException exception)
{
- throw new RuntimeException(exception);
+ throw new SidecarApiCallException("Unable to retrieve time skew information. clusterId=" + clusterId(), exception);
+ }
+
+ Instant remoteNow = Instant.ofEpochMilli(timeSkew.currentTime);
+ Duration allowedDuration = Duration.ofMinutes(timeSkew.allowableSkewInMinutes);
+ if (localNow.isBefore(remoteNow.minus(allowedDuration)) || localNow.isAfter(remoteNow.plus(allowedDuration)))
+ {
+ throw new TimeSkewTooLargeException(timeSkew.allowableSkewInMinutes, localNow, remoteNow, clusterId());
}
}
@@ -240,18 +265,6 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
}
}
- @NotNull
- protected ReplicationFactor getReplicationFactor()
- {
- String keyspaceSchema = getKeyspaceSchema(true);
- if (keyspaceSchema == null)
- {
- throw new RuntimeException(String.format("Could not retrieve keyspace schema information for keyspace %s",
- conf.keyspace));
- }
- return CqlUtils.extractReplicationFactor(keyspaceSchema, conf.keyspace);
- }
-
@Override
public String getKeyspaceSchema(boolean cached)
{
@@ -279,6 +292,30 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
}
}
+ @Override
+ public ReplicationFactor replicationFactor()
+ {
+ ReplicationFactor rf = replicationFactor;
+ if (rf != null)
+ {
+ return rf;
+ }
+
+ String keyspaceSchema = getKeyspaceSchema(true);
+ if (keyspaceSchema == null)
+ {
+ throw new RuntimeException("Could not retrieve keyspace schema information for keyspace " + conf.keyspace);
+ }
+ synchronized (this)
+ {
+ if (replicationFactor == null)
+ {
+ replicationFactor = CqlUtils.extractReplicationFactor(keyspaceSchema, conf.keyspace);
+ }
+ return replicationFactor;
+ }
+ }
+
@Override
public TokenRangeMapping<RingInstance> getTokenRangeMapping(boolean cached)
{
@@ -293,7 +330,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
// We can avoid synchronization here
if (topology != null)
{
- topology = getTokenRangeReplicas();
+ topology = getTokenRangeReplicasFromSidecar();
this.tokenRangeReplicas = topology;
return topology;
}
@@ -303,7 +340,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
{
try
{
- this.tokenRangeReplicas = getTokenRangeReplicas();
+ this.tokenRangeReplicas = getTokenRangeReplicasFromSidecar();
}
catch (Exception exception)
{
@@ -363,12 +400,11 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
return WriteAvailability.determineFromNodeState(instance.nodeState(), instance.nodeStatus());
}
- private TokenRangeMapping<RingInstance> getTokenRangeReplicas()
+ private TokenRangeMapping<RingInstance> getTokenRangeReplicasFromSidecar()
{
return TokenRangeMapping.create(this::getTokenRangesAndReplicaSets,
this::getPartitioner,
- this::getReplicationFactor,
- RingInstance::new);
+ metadata -> new RingInstance(metadata, clusterId()));
}
public String getVersionFromFeature()
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraSchemaInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraSchemaInfo.java
index 7c63320..526d9a7 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraSchemaInfo.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraSchemaInfo.java
@@ -21,15 +21,13 @@ package org.apache.cassandra.spark.bulkwriter;
import java.util.Set;
-import org.apache.cassandra.spark.data.CqlTable;
-
public class CassandraSchemaInfo implements SchemaInfo
{
private static final long serialVersionUID = -2327383232935001862L;
private final TableSchema tableSchema;
private final Set<String> userDefinedTypeStatements;
- public CassandraSchemaInfo(TableSchema tableSchema, Set<String> userDefinedTypeStatements, CqlTable cqlTable)
+ public CassandraSchemaInfo(TableSchema tableSchema, Set<String> userDefinedTypeStatements)
{
this.tableSchema = tableSchema;
this.userDefinedTypeStatements = userDefinedTypeStatements;
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java
index c6fe930..61cd694 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java
@@ -20,12 +20,16 @@
package org.apache.cassandra.spark.bulkwriter;
import java.io.Serializable;
-import java.util.List;
+import java.math.BigInteger;
import java.util.Map;
-import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse;
+import com.google.common.collect.Range;
+
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
+import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
+import org.apache.cassandra.spark.exception.SidecarApiCallException;
+import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
import org.apache.cassandra.spark.validation.StartupValidatable;
import org.jetbrains.annotations.Nullable;
@@ -46,7 +50,13 @@ public interface ClusterInfo extends StartupValidatable, Serializable
void checkBulkWriterIsEnabledOrThrow();
- TimeSkewResponse getTimeSkew(List<RingInstance> replicas);
+ /**
+ * Validate whether the time skew of the replicas of the range is acceptable
+ * @param range token range used to look up the relevant replicas
+ * @throws SidecarApiCallException when fails to retrieve time skew information
+ * @throws TimeSkewTooLargeException when the time skew has exceeded the allowance
+ */
+ void validateTimeSkew(Range<BigInteger> range) throws SidecarApiCallException, TimeSkewTooLargeException;
/**
* Return the keyspace schema string of the enclosing keyspace for bulk write in the cluster
@@ -55,6 +65,11 @@ public interface ClusterInfo extends StartupValidatable, Serializable
*/
String getKeyspaceSchema(boolean cached);
+ /**
+ * @return ReplicationFactor of the enclosing keyspace for bulk write in the cluster
+ */
+ ReplicationFactor replicationFactor();
+
CassandraContext getCassandraContext();
/**
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java
index 52432cb..7c774a1 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java
@@ -140,7 +140,7 @@ public class DirectStreamSession extends StreamSession<TransportContext.DirectDa
streamResult.setCommitResults(cr);
LOGGER.debug("StreamResult: {}", streamResult);
// Check consistency given the no. failures
- BulkWriteValidator.validateClOrFail(tokenRangeMapping, failureHandler, LOGGER, WRITE_PHASE, writerContext.job());
+ BulkWriteValidator.validateClOrFail(tokenRangeMapping, failureHandler, LOGGER, WRITE_PHASE, writerContext.job(), writerContext.cluster());
return streamResult;
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinator.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinator.java
index 228ed91..11d393b 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinator.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinator.java
@@ -113,7 +113,7 @@ public final class ImportCompletionCoordinator
onCancelJob.accept(cancelJobEvent);
};
this.cassandraTopologyMonitor = monitorCreator.apply(writerContext.cluster(), wrapped);
- this.replicationFactor = cassandraTopologyMonitor.initialTopology().replicationFactor();
+ this.replicationFactor = writerContext.cluster().replicationFactor();
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java
index 8e3c30a..2cd6abf 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java
@@ -23,9 +23,6 @@ import java.io.IOException;
import java.math.BigInteger;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.time.Duration;
-import java.time.Instant;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -49,7 +46,6 @@ import com.google.common.collect.Range;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse;
import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.bulkwriter.util.TaskContextUtils;
@@ -151,15 +147,10 @@ public class RecordWriter
initialTokenRangeMapping.allInstances().size(),
initialTokenRangeMapping.pendingInstances().size());
- Map<Range<BigInteger>, List<RingInstance>> initialTokenRangeInstances =
- taskTokenRangeMapping(initialTokenRangeMapping, taskTokenRange);
-
writeValidator.setPhase("Environment Validation");
writeValidator.validateClOrFail(initialTokenRangeMapping);
writeValidator.setPhase("UploadAndCommit");
-
- // for all replicas in this partition
- validateAcceptableTimeSkewOrThrow(new ArrayList<>(instancesFromMapping(initialTokenRangeInstances)));
+ writerContext.cluster().validateTimeSkew(taskTokenRange);
Iterator<Tuple2<DecoratedKey, Object[]>> dataIterator = new JavaInterruptibleIterator<>(taskContext, sourceIterator);
int partitionId = taskContext.partitionId();
@@ -361,26 +352,6 @@ public class RecordWriter
return writerContext.job().getTokenPartitioner().getTokenRange(taskContext.partitionId());
}
- private void validateAcceptableTimeSkewOrThrow(List<RingInstance> replicas)
- {
- if (replicas.isEmpty())
- {
- return;
- }
-
- TimeSkewResponse timeSkewResponse = writerContext.cluster().getTimeSkew(replicas);
- Instant localNow = Instant.now();
- Instant remoteNow = Instant.ofEpochMilli(timeSkewResponse.currentTime);
- Duration range = Duration.ofMinutes(timeSkewResponse.allowableSkewInMinutes);
- if (localNow.isBefore(remoteNow.minus(range)) || localNow.isAfter(remoteNow.plus(range)))
- {
- final String message = String.format("Time skew between Spark and Cassandra is too large. "
- + "Allowable skew is %d minutes. Spark executor time is %s, Cassandra instance time is %s",
- timeSkewResponse.allowableSkewInMinutes, localNow, remoteNow);
- throw new UnsupportedOperationException(message);
- }
- }
-
private void writeRow(Tuple2<DecoratedKey, Object[]> keyAndRowData,
Map<String, Object> valueMap,
int partitionId,
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
index f0edb71..362e04c 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
@@ -167,7 +167,7 @@ public class RingInstance implements CassandraInstance, Serializable
@Override
public String toString()
{
- return ringEntry.toString();
+ return "Cluster=" + clusterId + ',' + ringEntry.toString();
}
public RingEntry ringEntry()
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/blobupload/BlobStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/blobupload/BlobStreamSession.java
index dd568df..681b6e0 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/blobupload/BlobStreamSession.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/blobupload/BlobStreamSession.java
@@ -184,7 +184,7 @@ public class BlobStreamSession extends StreamSession<TransportContext.CloudStora
LOGGER.info("StreamResult: {}", streamResult);
// If the number of successful createSliceRequests cannot satisfy the configured consistency level,
// an exception is thrown and the task is failed. Spark might retry the task.
- BulkWriteValidator.validateClOrFail(tokenRangeMapping, failureHandler, LOGGER, WRITE_PHASE, writerContext.job());
+ BulkWriteValidator.validateClOrFail(tokenRangeMapping, failureHandler, LOGGER, WRITE_PHASE, writerContext.job(), writerContext.cluster());
return streamResult;
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraClusterInfoGroup.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraClusterInfoGroup.java
new file mode 100644
index 0000000..f3ff112
--- /dev/null
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraClusterInfoGroup.java
@@ -0,0 +1,269 @@
+/*
+ * 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.spark.bulkwriter.coordinatedwrite;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.bridge.CassandraVersionFeatures;
+import org.apache.cassandra.spark.bulkwriter.CassandraContext;
+import org.apache.cassandra.spark.bulkwriter.ClusterInfo;
+import org.apache.cassandra.spark.bulkwriter.RingInstance;
+import org.apache.cassandra.spark.bulkwriter.WriteAvailability;
+import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
+import org.apache.cassandra.spark.data.ReplicationFactor;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
+import org.apache.cassandra.spark.exception.SidecarApiCallException;
+import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+public class CassandraClusterInfoGroup implements ClusterInfo, MultiClusterInfoProvider
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(CassandraClusterInfoGroup.class);
+
+ private static final long serialVersionUID = 5337884321245616172L;
+
+ // immutable
+ private final List<ClusterInfo> clusterInfos;
+ private transient volatile Map<String, ClusterInfo> clusterInfoById;
+ private transient volatile TokenRangeMapping<RingInstance> consolidatedTokenRangeMapping;
+
+ public CassandraClusterInfoGroup(List<ClusterInfo> clusterInfos)
+ {
+ Preconditions.checkArgument(clusterInfos != null && !clusterInfos.isEmpty(),
+ "clusterInfos cannot be null or empty");
+ this.clusterInfos = Collections.unmodifiableList(clusterInfos);
+ buildClusterInfoById();
+ }
+
+ @Override
+ public void refreshClusterInfo()
+ {
+ runOnEach(ClusterInfo::refreshClusterInfo);
+ }
+
+ @Override
+ public TokenRangeMapping<RingInstance> getTokenRangeMapping(boolean cached)
+ {
+ if (clusterInfos.size() == 1)
+ {
+ return clusterInfos.get(0).getTokenRangeMapping(cached);
+ }
+
+ if (!cached || consolidatedTokenRangeMapping == null)
+ {
+ synchronized (this)
+ {
+ // return immediately if consolidatedTokenRangeMapping has been initialized and call-site asks for the cached value
+ if (cached && consolidatedTokenRangeMapping != null)
+ {
+ return consolidatedTokenRangeMapping;
+ }
+ Map<String, TokenRangeMapping<RingInstance>> aggregated = applyOnEach(c -> c.getTokenRangeMapping(cached));
+ consolidatedTokenRangeMapping = TokenRangeMapping.consolidate(new ArrayList<>(aggregated.values()));
+ }
+ }
+
+ return consolidatedTokenRangeMapping;
+ }
+
+ /**
+ * @return the lowest cassandra version among all clusters
+ */
+ @Override
+ public String getLowestCassandraVersion()
+ {
+ if (clusterInfos.size() == 1)
+ {
+ return clusterInfos.get(0).getLowestCassandraVersion();
+ }
+
+ Map<String, String> aggregated = applyOnEach(ClusterInfo::getLowestCassandraVersion);
+ List<CassandraVersionFeatures> versions = aggregated.values()
+ .stream()
+ .map(CassandraVersionFeatures::cassandraVersionFeaturesFromCassandraVersion)
+ .sorted()
+ .collect(Collectors.toList());
+ CassandraVersionFeatures first = versions.get(0);
+ CassandraVersionFeatures last = versions.get(versions.size() - 1);
+ Preconditions.checkState(first.getMajorVersion() == last.getMajorVersion(),
+ "Cluster versions are not compatible. lowest=%s and highest=%s",
+ first.getRawVersionString(), last.getRawVersionString());
+
+ return first.getRawVersionString();
+ }
+
+ @Override
+ public Map<RingInstance, WriteAvailability> clusterWriteAvailability()
+ {
+ if (clusterInfos.size() == 1)
+ {
+ return clusterInfos.get(0).clusterWriteAvailability();
+ }
+
+ Map<String, Map<RingInstance, WriteAvailability>> aggregated = applyOnEach(ClusterInfo::clusterWriteAvailability);
+ Map<RingInstance, WriteAvailability> consolidated = new HashMap<>();
+ aggregated.values().forEach(consolidated::putAll);
+ return consolidated;
+ }
+
+ @Override
+ public Partitioner getPartitioner()
+ {
+ Map<String, Partitioner> aggregated = applyOnEach(ClusterInfo::getPartitioner);
+ Set<Partitioner> partitioners = EnumSet.copyOf(aggregated.values());
+ if (partitioners.size() != 1)
+ {
+ throw new IllegalStateException("Clusters are not running with the same partitioner kind. Found partitioners: " + aggregated);
+ }
+
+ return partitioners.iterator().next();
+ }
+
+ @Override
+ public void checkBulkWriterIsEnabledOrThrow()
+ {
+ runOnEach(ClusterInfo::checkBulkWriterIsEnabledOrThrow);
+ }
+
+ @Override
+ public void validateTimeSkew(Range<BigInteger> range) throws SidecarApiCallException, TimeSkewTooLargeException
+ {
+ clusterInfos.forEach(ci -> ci.validateTimeSkew(range));
+ }
+
+ @Override
+ public String getKeyspaceSchema(boolean cached)
+ {
+ // All clusters that receive write should have the same keyspace schema. Therefore, return from the first cluster
+ // Note that the keyspace replication options can vary among the clusters. It is/should not be used when the correct ReplicationFactor is wanted.
+ // Instead, call the replicationFactor() method on the individual ClusterInfo
+ return clusterInfos.get(0).getKeyspaceSchema(cached);
+ }
+
+ @Override
+ public ReplicationFactor replicationFactor()
+ {
+ // Call the replicationFactor() method on the individual ClusterInfo
+ throw new UnsupportedOperationException("Not implemented in CassandraClusterInfoGroup");
+ }
+
+ @Override
+ public CassandraContext getCassandraContext()
+ {
+ // Call the getCassandraContext() method on the individual ClusterInfo
+ throw new UnsupportedOperationException("Not implemented in CassandraClusterInfoGroup");
+ }
+
+ @Override
+ public void startupValidate()
+ {
+ runOnEach(ClusterInfo::startupValidate);
+ }
+
+ @Override
+ public String clusterId()
+ {
+ return "ClusterInfoGroup: [" + String.join(", ", applyOnEach(ClusterInfo::clusterId).values()) + ']';
+ }
+
+ @Override
+ public List<ClusterInfo> clusters()
+ {
+ return clusterInfos;
+ }
+
+ @Nullable
+ @Override
+ public ClusterInfo cluster(@NotNull String clusterId)
+ {
+ if (clusterInfoById == null)
+ {
+ buildClusterInfoById();
+ }
+ return clusterInfoById.get(clusterId);
+ }
+
+ private synchronized void buildClusterInfoById()
+ {
+ if (clusterInfoById != null)
+ {
+ return;
+ }
+
+ clusterInfoById = clusterInfos.stream().collect(Collectors.toMap(ClusterInfo::clusterId, Function.identity()));
+ }
+
+ private void runOnEach(Consumer<ClusterInfo> action)
+ {
+ for (ClusterInfo clusterInfo : clusterInfos)
+ {
+ try
+ {
+ action.accept(clusterInfo);
+ }
+ catch (Throwable cause)
+ {
+ throw toRuntimeException(clusterInfo, cause);
+ }
+ }
+ }
+
+ private <T> Map<String, T> applyOnEach(Function<ClusterInfo, T> action)
+ {
+ // Preserve order with LinkedHashMap
+ Map<String, T> aggregated = new LinkedHashMap<>(clusterInfos.size());
+ for (ClusterInfo clusterInfo : clusterInfos)
+ {
+ try
+ {
+ // clusterId should not be null when there are multiple clusters
+ aggregated.put(clusterInfo.clusterId(), action.apply(clusterInfo));
+ }
+ catch (Throwable cause)
+ {
+ throw toRuntimeException(clusterInfo, cause);
+ }
+ }
+ return aggregated;
+ }
+
+ private RuntimeException toRuntimeException(ClusterInfo clusterInfo, Throwable cause)
+ {
+ LOGGER.error("Failed to perform action on cluster. cluster={}", clusterInfo.clusterId(), cause);
+ return new RuntimeException("Failed to perform action on cluster: " + clusterInfo.clusterId(), cause);
+ }
+}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraCoordinatedBulkWriterContext.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraCoordinatedBulkWriterContext.java
new file mode 100644
index 0000000..d605aea
--- /dev/null
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraCoordinatedBulkWriterContext.java
@@ -0,0 +1,98 @@
+/*
+ * 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.spark.bulkwriter.coordinatedwrite;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+
+import org.apache.cassandra.spark.bulkwriter.AbstractBulkWriterContext;
+import org.apache.cassandra.spark.bulkwriter.BulkSparkConf;
+import org.apache.cassandra.spark.bulkwriter.CassandraClusterInfo;
+import org.apache.cassandra.spark.bulkwriter.ClusterInfo;
+import org.apache.spark.sql.types.StructType;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * BulkWriterContext for coordinated write
+ * The context requires the coordinated-write configuration to be present.
+ */
+public class CassandraCoordinatedBulkWriterContext extends AbstractBulkWriterContext
+{
+ private static final long serialVersionUID = -2296507634642008675L;
+ private CassandraClusterInfoGroup clusterInfoGroup;
+
+ public CassandraCoordinatedBulkWriterContext(@NotNull BulkSparkConf conf,
+ @NotNull StructType structType,
+ int sparkDefaultParallelism)
+ {
+ super(conf, structType, sparkDefaultParallelism);
+ Preconditions.checkArgument(conf.isCoordinatedWriteConfigured(),
+ "Cannot create CassandraCoordinatedBulkWriterContext without CoordinatedWrite configuration");
+ }
+
+ @Override
+ protected ClusterInfo buildClusterInfo()
+ {
+ List<ClusterInfo> clusters = createClusterInfosForCoordinatedWrite(bulkSparkConf());
+ Preconditions.checkState(!clusters.isEmpty(), "No cluster info is built from the bulk writer option");
+ clusterInfoGroup = new CassandraClusterInfoGroup(clusters);
+ clusterInfoGroup.startupValidate();
+ return clusterInfoGroup;
+ }
+
+ @Override
+ protected void validateKeyspaceReplication()
+ {
+ BulkSparkConf conf = bulkSparkConf();
+ // no validation for non-local CL
+ if (!conf.consistencyLevel.isLocal())
+ {
+ return;
+ }
+ CoordinatedWriteConf coordinatedWriteConf = conf.coordinatedWriteConf();
+ // all specified clusters contain the corresponding replication DCs
+ coordinatedWriteConf.clusters().forEach((clusterId, clusterConf) -> {
+ // localDc is not empty and replication option contains localDc
+ String localDc = clusterConf.localDc();
+ ClusterInfo cluster = clusterInfoGroup.cluster(clusterId);
+ boolean isReplicatedToLocalDc = !StringUtils.isEmpty(localDc)
+ && cluster != null
+ && cluster.replicationFactor().getOptions().containsKey(localDc);
+ Preconditions.checkState(isReplicatedToLocalDc,
+ "Keyspace %s is not replicated on datacenter %s in %s",
+ conf.keyspace, localDc, clusterId);
+ });
+ }
+
+ static List<ClusterInfo> createClusterInfosForCoordinatedWrite(BulkSparkConf conf)
+ {
+ CoordinatedWriteConf coordinatedWriteConf = conf.coordinatedWriteConf();
+ Preconditions.checkArgument(coordinatedWriteConf != null,
+ "CoordinatedWriteConf must present for CassandraCoordinatedBulkWriterContext");
+ return coordinatedWriteConf.clusters()
+ .keySet()
+ .stream()
+ .map(clusterId -> new CassandraClusterInfo(conf, clusterId))
+ .collect(Collectors.toList());
+ }
+}
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/SidecarApiCallException.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/MultiClusterInfoProvider.java
similarity index 55%
copy from cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/SidecarApiCallException.java
copy to cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/MultiClusterInfoProvider.java
index d59c138..9faab9e 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/exception/SidecarApiCallException.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/MultiClusterInfoProvider.java
@@ -17,15 +17,29 @@
* under the License.
*/
-package org.apache.cassandra.spark.exception;
+package org.apache.cassandra.spark.bulkwriter.coordinatedwrite;
+
+import java.util.List;
+
+import org.apache.cassandra.spark.bulkwriter.ClusterInfo;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
- * Exception due to Cassandra Sidecar Api call failure
+ * Provider for multiple ClusterInfo and lookup
*/
-public class SidecarApiCallException extends AnalyticsException
+public interface MultiClusterInfoProvider
{
- public SidecarApiCallException(String message, Throwable cause)
- {
- super(message, cause);
- }
+ /**
+ * @return the list of ClusterInfo
+ */
+ List<ClusterInfo> clusters();
+
+ /**
+ * Look up a ClusterInfo based on clusterId
+ * @param clusterId cluster id
+ * @return the ClusterInfo associated with the clusterId, or null if not found
+ */
+ @Nullable
+ ClusterInfo cluster(@NotNull String clusterId);
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java
index fe2c36b..f821ba9 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java
@@ -155,7 +155,8 @@ public class ReplicaAwareFailureHandler<Instance extends CassandraInstance>
public synchronized List<ConsistencyFailurePerRange>
getFailedRanges(TokenRangeMapping<Instance> tokenRangeMapping,
ConsistencyLevel cl,
- @Nullable String localDC)
+ @Nullable String localDC,
+ ReplicationFactor replicationFactor)
{
Preconditions.checkArgument((cl.isLocal() && localDC != null) || (!cl.isLocal() && localDC == null),
"Not a valid pair of consistency level configuration. " +
@@ -180,7 +181,7 @@ public class ReplicaAwareFailureHandler<Instance extends CassandraInstance>
tokenRangeMapping.getWriteReplicasOfRange(range, localDC)
.forEach((subrange, liveAndDown) -> {
- if (!checkSubrange(cl, localDC, tokenRangeMapping.replicationFactor(), liveAndDown, failedReplicas))
+ if (!checkSubrange(cl, localDC, replicationFactor, liveAndDown, failedReplicas))
{
failedRanges.add(new ConsistencyFailurePerRange(subrange, errorMap));
}
@@ -213,11 +214,4 @@ public class ReplicaAwareFailureHandler<Instance extends CassandraInstance>
return cl.canBeSatisfied(succeededReplicas, pendingReplicas, replicationFactor, localDC);
}
-
- public boolean hasFailed(TokenRangeMapping<Instance> tokenRange,
- ConsistencyLevel cl,
- String localDC)
- {
- return !getFailedRanges(tokenRange, cl, localDC).isEmpty();
- }
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java
index 4159b7f..367c385 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java
@@ -46,8 +46,8 @@ import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse;
import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse.ReplicaInfo;
import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse.ReplicaMetadata;
import org.apache.cassandra.spark.common.model.CassandraInstance;
-import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
public class TokenRangeMapping<I extends CassandraInstance> implements Serializable
@@ -56,7 +56,6 @@ public class TokenRangeMapping<I extends CassandraInstance> implements Serializa
private static final Logger LOGGER = LoggerFactory.getLogger(TokenRangeMapping.class);
private final Partitioner partitioner;
- private final ReplicationFactor replicationFactor;
private final transient Set<I> allInstances;
private final transient Set<I> pendingInstances;
private final transient RangeMap<BigInteger, List<I>> instancesByTokenRange;
@@ -65,7 +64,6 @@ public class TokenRangeMapping<I extends CassandraInstance> implements Serializa
public static <I extends CassandraInstance>
TokenRangeMapping<I> create(Supplier<TokenRangeReplicasResponse> topologySupplier,
Supplier<Partitioner> partitionerSupplier,
- Supplier<ReplicationFactor> replicationFactorSupplier,
Function<ReplicaMetadata, I> instanceCreator)
{
TokenRangeReplicasResponse response = topologySupplier.get();
@@ -83,11 +81,39 @@ public class TokenRangeMapping<I extends CassandraInstance> implements Serializa
}
return new TokenRangeMapping<>(partitionerSupplier.get(),
- replicationFactorSupplier.get(),
tokenRangesByInstance,
allInstances);
}
+ /**
+ * Consolidate all TokenRangeMapping and produce a new TokenRangeMapping instance
+ * @param all list of TokenRangeMapping
+ * @return a consolidated TokenRangeMapping
+ * @param <I> CassandraInstance type
+ */
+ public static <I extends CassandraInstance>
+ TokenRangeMapping<I> consolidate(@NotNull List<TokenRangeMapping<I>> all)
+ {
+ Preconditions.checkArgument(!all.isEmpty(), "Cannot consolidate TokenRangeMapping from none");
+
+ if (all.size() == 1)
+ {
+ return all.get(0);
+ }
+
+ Set<Partitioner> partitioners = all.stream().map(t -> t.partitioner).collect(Collectors.toSet());
+ Preconditions.checkArgument(partitioners.size() == 1, "Multiple Partitioners found: " + partitioners);
+ Partitioner partitioner = all.get(0).partitioner;
+ Set<I> allInstances = new HashSet<>();
+ Multimap<I, Range<BigInteger>> tokenRangesByInstance = ArrayListMultimap.create();
+ for (TokenRangeMapping<I> topology : all)
+ {
+ allInstances.addAll(topology.allInstances);
+ tokenRangesByInstance.putAll(topology.getTokenRanges());
+ }
+ return new TokenRangeMapping<>(partitioner, tokenRangesByInstance, allInstances);
+ }
+
private static <I extends CassandraInstance>
Multimap<I, Range<BigInteger>> tokenRangesByInstance(List<ReplicaInfo> writeReplicas,
Map<String, I> instanceByIpAddress)
@@ -113,12 +139,10 @@ public class TokenRangeMapping<I extends CassandraInstance> implements Serializa
}
public TokenRangeMapping(Partitioner partitioner,
- ReplicationFactor replicationFactor,
Multimap<I, Range<BigInteger>> tokenRanges,
Set<I> allInstances)
{
this.partitioner = partitioner;
- this.replicationFactor = replicationFactor;
this.tokenRangeMap = tokenRanges;
this.allInstances = allInstances;
this.pendingInstances = allInstances.stream()
@@ -133,11 +157,6 @@ public class TokenRangeMapping<I extends CassandraInstance> implements Serializa
return partitioner;
}
- public ReplicationFactor replicationFactor()
- {
- return replicationFactor;
- }
-
/**
* Add a replica with given range to replicaMap (RangeMap pointing to replicas).
* <p>
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/CassandraDataSink.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/CassandraDataSink.java
index ad87277..bdd6bc1 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/CassandraDataSink.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/CassandraDataSink.java
@@ -22,12 +22,11 @@ package org.apache.cassandra.spark.sparksql;
import org.apache.cassandra.bridge.CassandraBridgeFactory;
import org.apache.cassandra.bridge.CassandraVersion;
import org.apache.cassandra.spark.bulkwriter.BulkWriterContext;
+import org.apache.cassandra.spark.bulkwriter.BulkWriterContextFactory;
import org.apache.cassandra.spark.bulkwriter.CassandraBulkSourceRelation;
-import org.apache.cassandra.spark.bulkwriter.CassandraBulkWriterContext;
import org.apache.cassandra.spark.bulkwriter.JobInfo;
import org.apache.cassandra.spark.bulkwriter.LoadNotSupportedException;
import org.apache.cassandra.spark.utils.ScalaConversionUtils;
-import org.apache.spark.SparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
@@ -35,7 +34,6 @@ import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.sources.BaseRelation;
import org.apache.spark.sql.sources.CreatableRelationProvider;
import org.apache.spark.sql.sources.DataSourceRegister;
-import org.apache.spark.sql.types.StructType;
import org.jetbrains.annotations.NotNull;
import scala.collection.immutable.Map;
@@ -71,9 +69,7 @@ public class CassandraDataSink implements DataSourceRegister, CreatableRelationP
switch (saveMode)
{
case Append:
- // Initialize the job group ID for later use if we need to cancel the job
- // TODO: Can we get a more descriptive "description" in here from the end user somehow?
- BulkWriterContext writerContext = createBulkWriterContext(
+ BulkWriterContext writerContext = factory().createBulkWriterContext(
sqlContext.sparkContext(),
ScalaConversionUtils.<String, String>mapAsJavaMap(parameters),
data.schema());
@@ -82,6 +78,8 @@ public class CassandraDataSink implements DataSourceRegister, CreatableRelationP
JobInfo jobInfo = writerContext.job();
String description = "Cassandra Bulk Load for table " + jobInfo.qualifiedTableName();
CassandraBulkSourceRelation relation = new CassandraBulkSourceRelation(writerContext, sqlContext);
+ // Initialize the job group ID for later use if we need to cancel the job
+ // TODO: Can we get a more descriptive "description" in here from the end user somehow?
sqlContext.sparkContext().setJobGroup(jobInfo.getId(), description, false);
relation.insert(data, false);
return relation;
@@ -103,10 +101,8 @@ public class CassandraDataSink implements DataSourceRegister, CreatableRelationP
}
@NotNull
- protected BulkWriterContext createBulkWriterContext(@NotNull SparkContext sparkContext,
- @NotNull java.util.Map<String, String> options,
- @NotNull StructType schema)
+ protected BulkWriterContextFactory factory()
{
- return CassandraBulkWriterContext.fromOptions(sparkContext, options, schema);
+ return BulkWriterContextFactory.INSTANCE;
}
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/bridge/CassandraVersionFeaturesTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/bridge/CassandraVersionFeaturesTest.java
index 842477d..d9fe0b2 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/bridge/CassandraVersionFeaturesTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/bridge/CassandraVersionFeaturesTest.java
@@ -58,5 +58,6 @@ public class CassandraVersionFeaturesTest
assertEquals(major, features.getMajorVersion(), "Wrong major version for " + version + ",");
assertEquals(minor, features.getMinorVersion(), "Wrong minor version for " + version + ",");
assertEquals(suffix, features.getSuffix(), "Wrong version suffix for " + version + ",");
+ assertEquals(version, features.getRawVersionString());
}
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidatorTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidatorTest.java
index 59d7705..ef6cb13 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidatorTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidatorTest.java
@@ -52,11 +52,11 @@ class BulkWriteValidatorTest
Map<String, String> replicationOptions = new HashMap<>();
replicationOptions.put("class", "SimpleStrategy");
replicationOptions.put("replication_factor", "3");
+ when(mockClusterInfo.replicationFactor()).thenReturn(new ReplicationFactor(replicationOptions));
TokenRangeMapping<RingInstance> topology = TokenRangeMapping.create(
() -> TokenRangeMappingUtils.mockSimpleTokenRangeReplicasResponse(10, 3),
() -> Partitioner.Murmur3Partitioner,
- () -> new ReplicationFactor(replicationOptions),
- RingInstance::new);
+ metadata -> new RingInstance(metadata, null));
when(mockClusterInfo.getTokenRangeMapping(anyBoolean())).thenReturn(topology);
Map<RingInstance, WriteAvailability> instanceAvailabilityMap = new HashMap<>(10);
for (RingInstance instance : topology.getTokenRanges().keySet())
@@ -77,7 +77,8 @@ class BulkWriteValidatorTest
when(mockJobInfo.jobKeepAliveMinutes()).thenReturn(-1);
when(mockWriterContext.job()).thenReturn(mockJobInfo);
- BulkWriteValidator writerValidator = new BulkWriteValidator(mockWriterContext, new ReplicaAwareFailureHandler<>(Partitioner.Murmur3Partitioner));
+ ReplicaAwareFailureHandler<RingInstance> failureHandler = new ReplicaAwareFailureHandler<>(Partitioner.Murmur3Partitioner);
+ BulkWriteValidator writerValidator = new BulkWriteValidator(mockWriterContext, failureHandler);
assertThatThrownBy(() -> writerValidator.validateClOrFail(topology))
.isExactlyInstanceOf(ConsistencyNotSatisfiedException.class)
.hasMessageContaining("Failed to write");
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkWriterContextFactoryTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkWriterContextFactoryTest.java
new file mode 100644
index 0000000..6543512
--- /dev/null
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkWriterContextFactoryTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.spark.bulkwriter;
+
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.spark.bulkwriter.coordinatedwrite.CassandraCoordinatedBulkWriterContext;
+import org.apache.cassandra.spark.common.stats.JobStatsPublisher;
+import org.apache.cassandra.spark.data.QualifiedTableName;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkContext;
+import org.apache.spark.sql.types.StructType;
+import org.jetbrains.annotations.NotNull;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class BulkWriterContextFactoryTest
+{
+ SparkContext sparkContext;
+ Map<String, String> options = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ TestBulkWriterContextFactory testFactory = new TestBulkWriterContextFactory();
+ StructType schema = mock(StructType.class);
+
+ @BeforeEach
+ void setup()
+ {
+ sparkContext = mock(SparkContext.class);
+ when(sparkContext.defaultParallelism()).thenReturn(1);
+ SparkConf conf = new SparkConf();
+ when(sparkContext.getConf()).thenReturn(conf);
+ options.put(WriterOptions.SIDECAR_CONTACT_POINTS.name(), "127.0.0.1");
+ options.put(WriterOptions.KEYSPACE.name(), "ks");
+ options.put(WriterOptions.TABLE.name(), "table");
+ options.put(WriterOptions.KEYSTORE_PASSWORD.name(), "dummy_password");
+ options.put(WriterOptions.KEYSTORE_PATH.name(), "dummy_path");
+ }
+
+ @Test
+ void testCreateBulkWriterContext()
+ {
+ assertThat(testFactory.createBulkWriterContext(sparkContext, options, schema)).isInstanceOf(CassandraBulkWriterContext.class);
+ }
+
+ @Test
+ void testCreateCoordinatedBulkWriterContext()
+ {
+ options.put(WriterOptions.DATA_TRANSPORT.name(), DataTransport.S3_COMPAT.name());
+ options.put(WriterOptions.COORDINATED_WRITE_CONFIG.name(), "{\"cluster1\":{\"sidecarContactPoints\":[\"instance-1:9999\"]}}");
+ assertThat(testFactory.createBulkWriterContext(sparkContext, options, schema)).isInstanceOf(CassandraCoordinatedBulkWriterContext.class);
+ }
+
+ static class TestBulkWriterContextFactory extends BulkWriterContextFactory
+ {
+ @NotNull
+ @Override
+ protected BulkWriterContext createCoordinatedBulkWriterContext(@NotNull BulkSparkConf conf, StructType schema, int sparkDefaultParallelism)
+ {
+ return setupMock(mock(CassandraCoordinatedBulkWriterContext.class));
+ }
+
+ @NotNull
+ @Override
+ protected BulkWriterContext createBulkWriterContext(@NotNull BulkSparkConf conf, StructType schema, int sparkDefaultParallelism)
+ {
+ return setupMock(mock(CassandraBulkWriterContext.class));
+ }
+
+ private BulkWriterContext setupMock(BulkWriterContext mock)
+ {
+ JobInfo job = mock(JobInfo.class);
+ when(job.getId()).thenReturn("id");
+ when(job.qualifiedTableName()).thenReturn(new QualifiedTableName("keyspace", "table"));
+ when(mock.job()).thenReturn(job);
+ when(mock.jobStats()).thenReturn(mock(JobStatsPublisher.class));
+ return mock;
+ }
+ }
+}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
new file mode 100644
index 0000000..846b897
--- /dev/null
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.spark.bulkwriter;
+
+import java.math.BigInteger;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Range;
+import org.junit.jupiter.api.Test;
+
+import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse;
+import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
+import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
+
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class CassandraClusterInfoTest
+{
+ @Test
+ void testTimeSkewAcceptable()
+ {
+ Instant localNow = Instant.now();
+ int allowanceMinutes = 10;
+ Instant remoteNow = localNow.plus(Duration.ofMinutes(1));
+ CassandraClusterInfo ci = mockClusterInfoForTimeSkewTest(allowanceMinutes, remoteNow);
+ assertThatNoException()
+ .describedAs("Acceptable time skew should validate without exception")
+ .isThrownBy(() -> ci.validateTimeSkewWithLocalNow(range(10, 20), localNow));
+ }
+
+ @Test
+ void testTimeSkewTooLarge()
+ {
+ Instant localNow = Instant.ofEpochMilli(1726604289530L);
+ int allowanceMinutes = 10;
+ Instant remoteNow = localNow.plus(Duration.ofMinutes(11)); // 11 > allowanceMinutes
+ CassandraClusterInfo ci = mockClusterInfoForTimeSkewTest(allowanceMinutes, remoteNow);
+ assertThatThrownBy(() -> ci.validateTimeSkewWithLocalNow(range(10, 20), localNow))
+ .describedAs("Time skew with too large a value should throw TimeSkewTooLargeException")
+ .isExactlyInstanceOf(TimeSkewTooLargeException.class)
+ .hasMessage("Time skew between Spark and Cassandra is too large. " +
+ "allowableSkewInMinutes=10, " +
+ "localTime=2024-09-17T20:18:09.530Z, " +
+ "remoteCassandraTime=2024-09-17T20:29:09.530Z, " +
+ "clusterId=null");
+ }
+
+ public static CassandraClusterInfo mockClusterInfoForTimeSkewTest(int allowanceMinutes, Instant remoteNow)
+ {
+ return new MockClusterInfoForTimeSkew(allowanceMinutes, remoteNow);
+ }
+
+ private static Range<BigInteger> range(long start, long end)
+ {
+ return Range.openClosed(BigInteger.valueOf(start), BigInteger.valueOf(end));
+ }
+
+ private static class MockClusterInfoForTimeSkew extends CassandraClusterInfo
+ {
+ private CassandraContext cassandraContext;
+
+ MockClusterInfoForTimeSkew(int allowanceMinutes, Instant remoteNow)
+ {
+ super(null);
+ mockCassandraContext(allowanceMinutes, remoteNow);
+ }
+
+ @Override
+ protected CassandraContext buildCassandraContext()
+ {
+ this.cassandraContext = mock(CassandraContext.class, RETURNS_DEEP_STUBS);
+ return cassandraContext;
+ }
+
+ @Override
+ public TokenRangeMapping<RingInstance> getTokenRangeMapping(boolean cached)
+ {
+ return TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("dc1", 3), 5);
+ }
+
+ private void mockCassandraContext(int allowanceMinutes, Instant remoteNow)
+ {
+ when(cassandraContext.getCluster()).thenReturn(Collections.emptySet());
+ TimeSkewResponse tsr = new TimeSkewResponse(remoteNow.toEpochMilli(), allowanceMinutes);
+ when(cassandraContext.getSidecarClient().timeSkew(any()))
+ .thenReturn(CompletableFuture.completedFuture(tsr));
+ when(cassandraContext.sidecarPort()).thenReturn(9043);
+ }
+ }
+}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java
index ae7a839..95af8d9 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java
@@ -42,11 +42,13 @@ import org.junit.jupiter.api.io.TempDir;
import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.common.model.CassandraInstance;
+import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.utils.DigestAlgorithm;
import org.apache.cassandra.spark.utils.XXHash32DigestAlgorithm;
import org.assertj.core.api.Assertions;
import org.jetbrains.annotations.NotNull;
+import static org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
@@ -83,8 +85,11 @@ public class DirectStreamSessionTest
{
digestAlgorithm = new XXHash32DigestAlgorithm();
range = Range.range(BigInteger.valueOf(101L), BoundType.CLOSED, BigInteger.valueOf(199L), BoundType.CLOSED);
- tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("DC1", 3), 12);
+ ImmutableMap<String, Integer> rfOptions = ImmutableMap.of("DC1", 3);
+ ReplicationFactor rf = new ReplicationFactor(NetworkTopologyStrategy, rfOptions);
+ tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(0, rfOptions, 12);
writerContext = getBulkWriterContext();
+ writerContext.setReplicationFactor(rf);
tableWriter = new MockTableWriter(folder);
transportContext = (TransportContext.DirectDataBulkWriterContext) writerContext.transportContext();
executor = new MockScheduledExecutorService();
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinatorTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinatorTest.java
index 578781b..7386c69 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinatorTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/ImportCompletionCoordinatorTest.java
@@ -59,12 +59,14 @@ import org.apache.cassandra.spark.bulkwriter.token.ConsistencyLevel;
import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.data.QualifiedTableName;
+import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
import org.apache.cassandra.spark.transports.storage.extensions.StorageTransportExtension;
import org.mockito.ArgumentCaptor;
import org.mockito.stubbing.Answer;
import static org.apache.cassandra.spark.bulkwriter.ImportCompletionCoordinator.estimateTimeoutNanos;
+import static org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -115,7 +117,10 @@ class ImportCompletionCoordinatorTest
CassandraContext mockCassandraContext = mock(CassandraContext.class);
when(mockClusterInfo.getCassandraContext()).thenReturn(mockCassandraContext);
- topology = TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("DC1", 3), TOTAL_INSTANCES);
+ ImmutableMap<String, Integer> rfOptions = ImmutableMap.of("DC1", 3);
+ ReplicationFactor rf = new ReplicationFactor(NetworkTopologyStrategy, rfOptions);
+ when(mockClusterInfo.replicationFactor()).thenReturn(rf);
+ topology = TokenRangeMappingUtils.buildTokenRangeMapping(0, rfOptions, TOTAL_INSTANCES);
when(mockClusterInfo.getTokenRangeMapping(anyBoolean())).thenReturn(topology);
when(mockWriterContext.job()).thenReturn(mockJobInfo);
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java
index 33b09a2..9fd90ad 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java
@@ -30,11 +30,9 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.function.BiFunction;
import java.util.function.Predicate;
-import java.util.function.Supplier;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
@@ -42,7 +40,6 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Range;
import org.apache.commons.lang3.tuple.Pair;
-import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CassandraBridgeFactory;
import org.apache.cassandra.bridge.CassandraVersion;
@@ -59,7 +56,10 @@ import org.apache.cassandra.spark.common.schema.ColumnTypes;
import org.apache.cassandra.spark.common.stats.JobStatsPublisher;
import org.apache.cassandra.spark.data.CqlField;
import org.apache.cassandra.spark.data.QualifiedTableName;
+import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
+import org.apache.cassandra.spark.exception.SidecarApiCallException;
+import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
import org.apache.cassandra.spark.validation.StartupValidator;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
@@ -84,6 +84,7 @@ public class MockBulkWriterContext implements BulkWriterContext, ClusterInfo, Jo
private ConsistencyLevel.CL consistencyLevel;
private int sstableDataSizeInMB = 128;
private CassandraBridge bridge = CassandraBridgeFactory.get(CassandraVersion.FOURZERO);
+ private TimeSkewTooLargeException timeSkewTooLargeException;
@Override
public void publish(Map<String, String> stats)
@@ -98,9 +99,6 @@ public class MockBulkWriterContext implements BulkWriterContext, ClusterInfo, Jo
public static final String DEFAULT_CASSANDRA_VERSION = "cassandra-4.0.2";
private final UUID jobId;
- private Supplier<Long> timeProvider = System::currentTimeMillis;
-
- private CountDownLatch uploadsLatch = new CountDownLatch(0);
private boolean skipClean = false;
public int refreshClusterInfoCallCount = 0; // CHECKSTYLE IGNORE: Public mutable field
private final Map<CassandraInstance, List<UploadRequest>> uploads = new ConcurrentHashMap<>();
@@ -113,8 +111,8 @@ public class MockBulkWriterContext implements BulkWriterContext, ClusterInfo, Jo
private final TokenPartitioner tokenPartitioner;
private final String cassandraVersion;
private CommitResultSupplier crSupplier = (uuids, dc) -> new DirectDataTransferApi.RemoteCommitResult(true, Collections.emptyList(), uuids, null);
-
private Predicate<CassandraInstance> uploadRequestConsumer = instance -> true;
+ private ReplicationFactor replicationFactor;
public MockBulkWriterContext(TokenRangeMapping<RingInstance> tokenRangeMapping)
{
@@ -168,42 +166,41 @@ public class MockBulkWriterContext implements BulkWriterContext, ClusterInfo, Jo
this.jobId = java.util.UUID.randomUUID();
}
- public Supplier<Long> getTimeProvider()
- {
- return timeProvider;
- }
-
- public void setTimeProvider(Supplier<Long> timeProvider)
+ @Override
+ public void shutdown()
{
- this.timeProvider = timeProvider;
}
- public CountDownLatch getUploadsLatch()
+ public void setTimeSkewTooLargeException(TimeSkewTooLargeException exception)
{
- return uploadsLatch;
+ this.timeSkewTooLargeException = exception;
}
- public void setUploadsLatch(CountDownLatch uploadsLatch)
+ @Override
+ public void validateTimeSkew(Range<BigInteger> range) throws SidecarApiCallException, TimeSkewTooLargeException
{
- this.uploadsLatch = uploadsLatch;
+ if (timeSkewTooLargeException != null)
+ {
+ throw timeSkewTooLargeException;
+ }
}
@Override
- public void shutdown()
+ public String getKeyspaceSchema(boolean cached)
{
+ // TODO: Fix me
+ throw new UnsupportedOperationException();
}
@Override
- public TimeSkewResponse getTimeSkew(List<RingInstance> replicas)
+ public ReplicationFactor replicationFactor()
{
- return new TimeSkewResponse(timeProvider.get(), 60);
+ return replicationFactor;
}
- @Override
- public String getKeyspaceSchema(boolean cached)
+ public void setReplicationFactor(ReplicationFactor replicationFactor)
{
- // TODO: Fix me
- throw new UnsupportedOperationException();
+ this.replicationFactor = replicationFactor;
}
@Override
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
index 9064766..38af194 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
@@ -21,6 +21,8 @@ package org.apache.cassandra.spark.bulkwriter;
import java.math.BigInteger;
import java.nio.file.Path;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
@@ -46,6 +48,8 @@ import org.apache.cassandra.spark.bulkwriter.token.ConsistencyLevel;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.common.model.CassandraInstance;
import org.apache.cassandra.spark.data.CqlField;
+import org.apache.cassandra.spark.data.ReplicationFactor;
+import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
import org.apache.cassandra.spark.utils.DigestAlgorithm;
import org.apache.cassandra.spark.utils.XXHash32DigestAlgorithm;
import org.apache.spark.sql.types.DataType;
@@ -59,11 +63,11 @@ import static org.apache.cassandra.spark.bulkwriter.SqlToCqlTypeConverter.DATE;
import static org.apache.cassandra.spark.bulkwriter.SqlToCqlTypeConverter.INT;
import static org.apache.cassandra.spark.bulkwriter.SqlToCqlTypeConverter.VARCHAR;
import static org.apache.cassandra.spark.bulkwriter.TableSchemaTestCommon.mockCqlType;
+import static org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.startsWith;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -100,9 +104,12 @@ class RecordWriterTest
{
digestAlgorithm = new XXHash32DigestAlgorithm();
tw = new MockTableWriter(folder.getRoot());
- tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(100000, ImmutableMap.of("DC1", 3), 12);
+ ImmutableMap<String, Integer> rfOption = ImmutableMap.of("DC1", 3);
+ ReplicationFactor rf = new ReplicationFactor(NetworkTopologyStrategy, rfOption);
+ tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(100000, rfOption, 12);
writerContext = new MockBulkWriterContext(tokenRangeMapping);
writerContext.setSstableDataSizeInMB(1); // defaults to the minimum sstable data size allowed to set
+ writerContext.setReplicationFactor(rf);
tc = new TestTaskContext();
range = writerContext.job().getTokenPartitioner().getTokenRange(tc.partitionId());
tokenizer = new Tokenizer(writerContext);
@@ -387,26 +394,18 @@ class RecordWriterTest
@Test
void testBadTimeSkewFails()
{
- // Mock context returns a 60-minute allowable time skew, so we use something just outside the limits
- long sixtyOneMinutesInMillis = TimeUnit.MINUTES.toMillis(61);
+ Instant localNow = Instant.ofEpochMilli(1726604289530L);
+ Instant remoteNow = localNow.minus(2, ChronoUnit.HOURS);
rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc,
(wc, path, dp, pid) -> new SortedSSTableWriter(tw, folder, digestAlgorithm, pid));
- writerContext.setTimeProvider(() -> System.currentTimeMillis() - sixtyOneMinutesInMillis);
+ writerContext.setTimeSkewTooLargeException(new TimeSkewTooLargeException(60, localNow, remoteNow, "test-cluster"));
Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData();
- RuntimeException ex = assertThrows(RuntimeException.class, () -> rw.write(data));
- assertThat(ex.getMessage(), startsWith("Time skew between Spark and Cassandra is too large. Allowable skew is 60 minutes. Spark executor time is "));
- }
-
- @Test
- void testTimeSkewWithinLimitsSucceeds()
- {
- // Mock context returns a 60-minute allowable time skew, so we use something just inside the limits
- long fiftyNineMinutesInMillis = TimeUnit.MINUTES.toMillis(59);
- long remoteTime = System.currentTimeMillis() - fiftyNineMinutesInMillis;
- rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc, SortedSSTableWriter::new);
- writerContext.setTimeProvider(() -> remoteTime); // Return a very low "current time" to make sure we fail if skew is too bad
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData();
- rw.write(data);
+ TimeSkewTooLargeException ex = assertThrows(TimeSkewTooLargeException.class, () -> rw.write(data));
+ assertEquals(ex.getMessage(), "Time skew between Spark and Cassandra is too large. " +
+ "allowableSkewInMinutes=60, " +
+ "localTime=2024-09-17T20:18:09.530Z, " +
+ "remoteCassandraTime=2024-09-17T18:18:09.530Z, " +
+ "clusterId=test-cluster");
}
private void validateSuccessfulWrite(MockBulkWriterContext writerContext,
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
index 0846e8f..022b44e 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
@@ -233,7 +233,6 @@ public class RingInstanceTest
ntsOptions(new String[]{DATACENTER_1 }, new int[]{3 }));
Multimap<RingInstance, Range<BigInteger>> tokenRanges = TokenRangeMappingUtils.setupTokenRangeMap(partitioner, repFactor, instances);
TokenRangeMapping<RingInstance> tokenRange = new TokenRangeMapping<>(partitioner,
- repFactor,
tokenRanges,
Collections.emptySet());
@@ -243,7 +242,7 @@ public class RingInstanceTest
replicationFactor3.addFailure(Range.openClosed(tokens[0].add(BigInteger.ONE),
tokens[0].add(BigInteger.valueOf(2L))), instance2, "Failure 2");
- assertTrue(replicationFactor3.getFailedRanges(tokenRange, CL.LOCAL_QUORUM, DATACENTER_1).isEmpty());
+ assertTrue(replicationFactor3.getFailedRanges(tokenRange, CL.LOCAL_QUORUM, DATACENTER_1, repFactor).isEmpty());
}
@NotNull
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java
index 8be92d2..5b0b927 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java
@@ -45,10 +45,12 @@ import org.apache.cassandra.spark.bulkwriter.token.ConsistencyLevel;
import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.common.model.CassandraInstance;
+import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.exception.ConsistencyNotSatisfiedException;
import org.apache.cassandra.spark.utils.DigestAlgorithm;
import org.apache.cassandra.spark.utils.XXHash32DigestAlgorithm;
+import static org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
@@ -64,8 +66,9 @@ public class StreamSessionConsistencyTest
// range [101, 199] is replicated to those instances. i1 has token 0/1, which are before the range
private static final List<String> EXPECTED_INSTANCES = ImmutableList.of("DC1-i2", "DC1-i3", "DC1-i4", "DC2-i2", "DC2-i3", "DC2-i4");
private static final Range<BigInteger> RANGE = Range.range(BigInteger.valueOf(101L), BoundType.CLOSED, BigInteger.valueOf(199L), BoundType.CLOSED);
+ private static final ImmutableMap<String, Integer> rfOptions = ImmutableMap.of("DC1", 3, "DC2", 3);
private static final TokenRangeMapping<RingInstance> TOKEN_RANGE_MAPPING =
- TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("DC1", 3, "DC2", 3), 6);
+ TokenRangeMappingUtils.buildTokenRangeMapping(0, rfOptions, 6);
private static final Map<String, Object> COLUMN_BIND_VALUES = ImmutableMap.of("id", 0, "date", 1, "course", "course", "marks", 2);
@TempDir
@@ -90,6 +93,7 @@ public class StreamSessionConsistencyTest
digestAlgorithm = new XXHash32DigestAlgorithm();
tableWriter = new MockTableWriter(folder);
writerContext = new MockBulkWriterContext(TOKEN_RANGE_MAPPING, "cassandra-4.0.0", consistencyLevel);
+ writerContext.setReplicationFactor(new ReplicationFactor(NetworkTopologyStrategy, rfOptions));
transportContext = (TransportContext.DirectDataBulkWriterContext) writerContext.transportContext();
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
index 202d3a3..6c1c1f2 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
@@ -81,7 +81,6 @@ public final class TokenRangeMappingUtils
Multimap<RingInstance, Range<BigInteger>> tokenRanges = setupTokenRangeMap(Partitioner.Murmur3Partitioner, replicationFactor, instances);
return new TokenRangeMapping<>(Partitioner.Murmur3Partitioner,
- replicationFactor,
tokenRanges,
new HashSet<>(instances));
}
@@ -118,7 +117,6 @@ public final class TokenRangeMappingUtils
ReplicationFactor replicationFactor = getReplicationFactor(rfByDC);
Multimap<RingInstance, Range<BigInteger>> tokenRanges = setupTokenRangeMap(Partitioner.Murmur3Partitioner, replicationFactor, instances);
return new TokenRangeMapping<>(Partitioner.Murmur3Partitioner,
- replicationFactor,
tokenRanges,
new HashSet<>(instances));
}
@@ -202,7 +200,11 @@ public final class TokenRangeMappingUtils
public static TokenRangeReplicasResponse mockSimpleTokenRangeReplicasResponse(int instancesCount, int replicationFactor)
{
- long startToken = 0;
+ return mockSimpleTokenRangeReplicasResponse(0L, instancesCount, replicationFactor);
+ }
+
+ public static TokenRangeReplicasResponse mockSimpleTokenRangeReplicasResponse(long startToken, int instancesCount, int replicationFactor)
+ {
long rangeLength = 100;
List<ReplicaInfo> replicaInfoList = new ArrayList<>(instancesCount);
Map<String, ReplicaMetadata> replicaMetadata = new HashMap<>(instancesCount);
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraClusterInfoGroupTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraClusterInfoGroupTest.java
new file mode 100644
index 0000000..f003b90
--- /dev/null
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/coordinatedwrite/CassandraClusterInfoGroupTest.java
@@ -0,0 +1,258 @@
+/*
+ * 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.spark.bulkwriter.coordinatedwrite;
+
+import java.math.BigInteger;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Range;
+import org.junit.jupiter.api.Test;
+
+import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse;
+import org.apache.cassandra.spark.bulkwriter.CassandraClusterInfo;
+import org.apache.cassandra.spark.bulkwriter.CassandraClusterInfoTest;
+import org.apache.cassandra.spark.bulkwriter.ClusterInfo;
+import org.apache.cassandra.spark.bulkwriter.RingInstance;
+import org.apache.cassandra.spark.bulkwriter.TokenRangeMappingUtils;
+import org.apache.cassandra.spark.bulkwriter.WriteAvailability;
+import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
+import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+class CassandraClusterInfoGroupTest
+{
+ @Test
+ void testCreateGroupFailWithEmptyList()
+ {
+ assertThatThrownBy(() -> new CassandraClusterInfoGroup(null))
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessage("clusterInfos cannot be null or empty");
+
+ assertThatThrownBy(() -> new CassandraClusterInfoGroup(Collections.emptyList()))
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessage("clusterInfos cannot be null or empty");
+ }
+
+ @Test
+ void testLookupCluster()
+ {
+ CassandraClusterInfoGroup group = mockClusterGroup(2, index -> mockClusterInfo("cluster" + index));
+ List<ClusterInfo> clusters = group.clusters();
+ assertThat(clusters).hasSize(2);
+ assertThat(group.cluster("cluster0")).isSameAs(clusters.get(0));
+ assertThat(group.cluster("cluster1")).isSameAs(clusters.get(1));
+ assertThat(group.cluster("cluster2")).isNull();
+ }
+
+ @Test
+ void testClusterId()
+ {
+ CassandraClusterInfoGroup group = mockClusterGroup(2, index -> mockClusterInfo("cluster" + index));
+ assertThat(group.clusterId()).isEqualTo("ClusterInfoGroup: [cluster0, cluster1]");
+
+ group = mockClusterGroup(1, index -> mockClusterInfo("cluster" + index));
+ assertThat(group.clusterId()).isEqualTo("ClusterInfoGroup: [cluster0]");
+ }
+
+ @Test
+ void testDelegationOfSingleCluster()
+ {
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster0");
+ TokenRangeReplicasResponse response = TokenRangeMappingUtils.mockSimpleTokenRangeReplicasResponse(10, 3);
+ TokenRangeMapping<RingInstance> expectedTokenRangeMapping = TokenRangeMapping.create(() -> response,
+ () -> Partitioner.Murmur3Partitioner,
+ RingInstance::new);
+ when(clusterInfo.getTokenRangeMapping(anyBoolean())).thenReturn(expectedTokenRangeMapping);
+ when(clusterInfo.getLowestCassandraVersion()).thenReturn("lowestCassandraVersion");
+ when(clusterInfo.clusterWriteAvailability()).thenReturn(Collections.emptyMap());
+ CassandraClusterInfoGroup group = mockClusterGroup(1, index -> clusterInfo);
+ // Since there is a single clusterInfo in the group. It behaves as a simple delegation to the sole clusterInfo
+ assertThat(group.clusterWriteAvailability()).isSameAs(clusterInfo.clusterWriteAvailability());
+ assertThat(group.getLowestCassandraVersion()).isSameAs(clusterInfo.getLowestCassandraVersion());
+ assertThat(group.getTokenRangeMapping(true)).isSameAs(clusterInfo.getTokenRangeMapping(true));
+ }
+
+ @Test
+ void testAggregatePartitioner()
+ {
+ CassandraClusterInfoGroup invalidGroup = mockClusterGroup(2, index -> {
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster" + index);
+ Partitioner partitioner = (index % 2 == 0) ? Partitioner.Murmur3Partitioner : Partitioner.RandomPartitioner;
+ when(clusterInfo.getPartitioner()).thenReturn(partitioner);
+ return clusterInfo;
+ });
+ assertThatThrownBy(invalidGroup::getPartitioner)
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessage("Clusters are not running with the same partitioner kind. Found partitioners: " +
+ "{cluster0=org.apache.cassandra.dht.Murmur3Partitioner, cluster1=org.apache.cassandra.dht.RandomPartitioner}");
+
+ CassandraClusterInfoGroup goodGroup = mockClusterGroup(2, index -> {
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster" + index);
+ when(clusterInfo.getPartitioner()).thenReturn(Partitioner.Murmur3Partitioner);
+ return clusterInfo;
+ });
+ assertThat(goodGroup.getPartitioner()).isEqualTo(Partitioner.Murmur3Partitioner);
+ }
+
+ @Test
+ void testAggregateWriteAvailability()
+ {
+ Map<RingInstance, WriteAvailability> cluster1Availability = ImmutableMap.of(mock(RingInstance.class), WriteAvailability.AVAILABLE);
+ Map<RingInstance, WriteAvailability> cluster2Availability = ImmutableMap.of(mock(RingInstance.class), WriteAvailability.UNAVAILABLE_DOWN);
+ CassandraClusterInfoGroup group = mockClusterGroup(2, index -> {
+ Map<RingInstance, WriteAvailability> availability = (index % 2 == 0) ? cluster1Availability : cluster2Availability;
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster" + index);
+ when(clusterInfo.clusterWriteAvailability()).thenReturn(availability);
+ return clusterInfo;
+ });
+ assertThat(group.clusterWriteAvailability())
+ .describedAs("clusterWriteAvailability retrieved from group contains entries from both clusters")
+ .hasSize(2)
+ .containsValues(WriteAvailability.AVAILABLE, WriteAvailability.UNAVAILABLE_DOWN);
+ }
+
+ @Test
+ void testAggregateLowestCassandraVersion()
+ {
+ CassandraClusterInfoGroup goodGroup = mockClusterGroup(2, index -> {
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster" + index);
+ when(clusterInfo.getLowestCassandraVersion()).thenReturn("4.0." + index);
+ return clusterInfo;
+ });
+ assertThat(goodGroup.getLowestCassandraVersion()).isEqualTo("4.0.0");
+ }
+
+ @Test
+ void testAggregateLowestCassandraVersionFailDueToDifference()
+ {
+ CassandraClusterInfoGroup badGroup = mockClusterGroup(2, index -> {
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster" + index);
+ when(clusterInfo.getLowestCassandraVersion()).thenReturn((4 + index) + ".0.0");
+ return clusterInfo;
+ });
+ assertThatThrownBy(badGroup::getLowestCassandraVersion)
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessage("Cluster versions are not compatible. lowest=4.0.0 and highest=5.0.0");
+ }
+
+ @Test
+ void testCheckBulkWriterIsEnabledOrThrow()
+ {
+ for (int i = 0; i < 2; i++)
+ {
+ int notEnabledClusterIndex = i;
+ CassandraClusterInfoGroup badGroup = mockClusterGroup(2, index -> {
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster" + index);
+ if (index == notEnabledClusterIndex)
+ {
+ doThrow(new RuntimeException("not enabled")).when(clusterInfo).checkBulkWriterIsEnabledOrThrow();
+ }
+ return clusterInfo;
+ });
+
+ assertThatThrownBy(badGroup::checkBulkWriterIsEnabledOrThrow)
+ .isExactlyInstanceOf(RuntimeException.class)
+ .hasMessage("Failed to perform action on cluster: cluster" + notEnabledClusterIndex)
+ .hasRootCauseMessage("not enabled");
+ }
+ }
+
+ @Test
+ void testAggregateTokenRangeMapping()
+ {
+ TokenRangeMapping<RingInstance> topology1 = TokenRangeMappingUtils.buildTokenRangeMapping(0,
+ ImmutableMap.of("dc1", 3),
+ 5);
+ TokenRangeMapping<RingInstance> topology2 = TokenRangeMappingUtils.buildTokenRangeMapping(10,
+ ImmutableMap.of("dc1", 3),
+ 8);
+ CassandraClusterInfoGroup group = mockClusterGroup(2, index -> {
+ TokenRangeMapping<RingInstance> topology = index == 0 ? topology1 : topology2;
+ CassandraClusterInfo clusterInfo = mockClusterInfo("cluster" + index);
+ when(clusterInfo.getTokenRangeMapping(anyBoolean())).thenReturn(topology);
+ return clusterInfo;
+ });
+ TokenRangeMapping<RingInstance> topology = group.getTokenRangeMapping(false);
+ TokenRangeMapping<RingInstance> expected = TokenRangeMapping.consolidate(Arrays.asList(topology1, topology2));
+ assertThat(topology).isEqualTo(expected);
+ TokenRangeMapping<RingInstance> cachedTopology = group.getTokenRangeMapping(true);
+ assertThat(cachedTopology).isSameAs(topology);
+ }
+
+ @Test
+ void testTimeSkewTooLarge()
+ {
+ for (int i = 0; i < 2; i++)
+ {
+ int clusterIndexWithLargeTimeSkew = i;
+ Instant remoteNow = Instant.ofEpochMilli(1726604289530L); // time in the past that guarantees to exceed the skew allowance
+ CassandraClusterInfoGroup group = mockClusterGroup(2, index -> {
+ if (index == clusterIndexWithLargeTimeSkew)
+ {
+ CassandraClusterInfo ci = spy(CassandraClusterInfoTest.mockClusterInfoForTimeSkewTest(10, remoteNow));
+ when(ci.clusterId()).thenReturn("cluster" + index);
+ return ci;
+ }
+ else
+ {
+ return mockClusterInfo("cluster" + index);
+ }
+ });
+
+ assertThatThrownBy(() -> group.validateTimeSkew(Range.openClosed(BigInteger.valueOf(10), BigInteger.valueOf(20))))
+ .isExactlyInstanceOf(TimeSkewTooLargeException.class)
+ .hasMessageContaining("Time skew between Spark and Cassandra is too large. ")
+ .hasMessageContaining("allowableSkewInMinutes=10, ")
+ .hasMessageContaining("remoteCassandraTime=2024-09-17T20:18:09.530Z, ")
+ .hasMessageContaining("clusterId=cluster" + clusterIndexWithLargeTimeSkew);
+ }
+
+ }
+
+ private CassandraClusterInfoGroup mockClusterGroup(int size,
+ Function<Integer, CassandraClusterInfo> clusterInfoCreator)
+ {
+ List<ClusterInfo> clusterInfos = IntStream.range(0, size).boxed().map(clusterInfoCreator).collect(Collectors.toList());
+ return new CassandraClusterInfoGroup(clusterInfos);
+ }
+
+ private CassandraClusterInfo mockClusterInfo(String clusterId)
+ {
+ CassandraClusterInfo clusterInfo = mock(CassandraClusterInfo.class);
+ when(clusterInfo.clusterId()).thenReturn(clusterId);
+ return clusterInfo;
+ }
+}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMappingTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMappingTest.java
index 3665121..b80538d 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMappingTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMappingTest.java
@@ -19,9 +19,13 @@
package org.apache.cassandra.spark.bulkwriter.token;
+import java.math.BigInteger;
+import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import com.google.common.collect.Range;
import org.junit.jupiter.api.Test;
import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse;
@@ -29,7 +33,6 @@ import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse.Re
import org.apache.cassandra.spark.bulkwriter.RingInstance;
import org.apache.cassandra.spark.bulkwriter.TokenRangeMappingUtils;
import org.apache.cassandra.spark.common.model.NodeState;
-import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
import static org.assertj.core.api.Assertions.assertThat;
@@ -37,18 +40,11 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
class TokenRangeMappingTest
{
- private final ReplicationFactor rf = new ReplicationFactor(new HashMap<String, String>()
- {{
- put("class", "SimpleStrategy");
- put("replication_factor", "3");
- }});
-
@Test
void testCreateTokenRangeMapping()
{
TokenRangeMapping<RingInstance> topology = createTestMapping(10);
assertThat(topology.partitioner()).isEqualTo(Partitioner.Murmur3Partitioner);
- assertThat(topology.replicationFactor()).isEqualTo(rf);
}
@Test
@@ -75,7 +71,7 @@ class TokenRangeMappingTest
});
assertThatThrownBy(() -> TokenRangeMapping.create(() -> invalidResponse,
() -> Partitioner.Murmur3Partitioner,
- () -> rf, RingInstance::new))
+ RingInstance::new))
.isExactlyInstanceOf(RuntimeException.class)
.hasMessage("No metadata found for instance: " + invalidInstance);
}
@@ -105,16 +101,145 @@ class TokenRangeMappingTest
}
TokenRangeMapping<RingInstance> topology = TokenRangeMapping.create(() -> response,
() -> Partitioner.Murmur3Partitioner,
- () -> rf, RingInstance::new);
+ RingInstance::new);
assertThat(topology.pendingInstances()).hasSize(pendingCount);
}
+ @Test
+ void testConsolidateListOfTokenRangeMappingFails()
+ {
+ assertThatThrownBy(() -> TokenRangeMapping.consolidate(Collections.emptyList()))
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot consolidate TokenRangeMapping from none");
+
+ TokenRangeMapping<RingInstance> topologyWithMurmur3 = createTestMapping(5, Partitioner.Murmur3Partitioner);
+ TokenRangeMapping<RingInstance> topologyWithRandom = createTestMapping(5, Partitioner.RandomPartitioner);
+ assertThatThrownBy(() -> TokenRangeMapping.consolidate(Arrays.asList(topologyWithMurmur3, topologyWithRandom)))
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Multiple Partitioners found: ")
+ .hasMessageContaining(Partitioner.Murmur3Partitioner.toString())
+ .hasMessageContaining(Partitioner.RandomPartitioner.toString());
+ }
+
+ @Test
+ void testConsolidateSingleTokenRangeMapping()
+ {
+ // Consolidate a single TokenRangeMapping
+ TokenRangeMapping<RingInstance> topology = createTestMapping(5);
+ TokenRangeMapping<RingInstance> consolidated = TokenRangeMapping.consolidate(Collections.singletonList(topology));
+ assertThat(consolidated).isSameAs(topology);
+ }
+
+ @Test
+ void testConsolidateFromTokenRangeMappingsWithSameRingLayout()
+ {
+ TokenRangeMapping<RingInstance> topology1 = createTestMapping(0, 5, Partitioner.Murmur3Partitioner, "cluster1");
+ TokenRangeMapping<RingInstance> topology2 = createTestMapping(0, 5, Partitioner.Murmur3Partitioner, "cluster2");
+ TokenRangeMapping<RingInstance> consolidated = TokenRangeMapping.consolidate(Arrays.asList(topology1, topology2));
+ assertThat(consolidated.allInstances())
+ .describedAs("Consolidated allInstances contains instances from both topologies")
+ .hasSize(10)
+ .containsAll(topology1.allInstances())
+ .containsAll(topology2.allInstances());
+ assertThat(consolidated.pendingInstances()).isEmpty();
+ assertThat(consolidated.getTokenRanges().asMap())
+ .describedAs("Consolidated TokenRanges contains instances and ranges from both topologies")
+ .hasSize(10)
+ .containsAllEntriesOf(topology1.getTokenRanges().asMap())
+ .containsAllEntriesOf(topology2.getTokenRanges().asMap());
+ assertThat(consolidated.getRangeMap().asMapOfRanges())
+ .describedAs("Consolidated RangeMap has the same key set (ranges) as both topologies because the range layouts are the same")
+ .hasSize(7)
+ .containsOnlyKeys(topology1.getRangeMap().asMapOfRanges().keySet())
+ .containsOnlyKeys(topology2.getRangeMap().asMapOfRanges().keySet());
+ // the instances per range in the consolidated should contain the instances of the same range from both topologies
+ consolidated.getRangeMap().asMapOfRanges().forEach((range, instances) -> {
+ if (instances.isEmpty())
+ {
+ return;
+ }
+ assertThat(instances)
+ .containsAll(topology1.getRangeMap().asMapOfRanges().get(range))
+ .containsAll(topology2.getRangeMap().asMapOfRanges().get(range));
+ });
+ }
+
+ @Test
+ void testConsolidateFromTokenRangeMappingsWithDistinctRingLayout()
+ {
+ TokenRangeMapping<RingInstance> topology1 = createTestMapping(0, 5, Partitioner.Murmur3Partitioner, "cluster1");
+ TokenRangeMapping<RingInstance> topology2 = createTestMapping(1, 5, Partitioner.Murmur3Partitioner, "cluster2");
+ TokenRangeMapping<RingInstance> consolidated = TokenRangeMapping.consolidate(Arrays.asList(topology1, topology2));
+ assertThat(consolidated.allInstances())
+ .describedAs("Consolidated allInstances contains instances from both topologies")
+ .hasSize(10)
+ .containsAll(topology1.allInstances())
+ .containsAll(topology2.allInstances());
+ assertThat(consolidated.pendingInstances()).isEmpty();
+ assertThat(consolidated.getTokenRanges().asMap())
+ .describedAs("Consolidated TokenRanges contains instances and ranges from both topologies")
+ .hasSize(10)
+ .containsAllEntriesOf(topology1.getTokenRanges().asMap())
+ .containsAllEntriesOf(topology2.getTokenRanges().asMap());
+ List<Range<BigInteger>> expectedConsolidatedRanges = Arrays.asList(
+ range(Long.MIN_VALUE, 0),
+ range(0, 1), range(1, 100),
+ range(100, 101), range(101, 200),
+ range(200, 201), range(201, 300),
+ range(300, 301), range(301, 400),
+ range(400, 401), range(401, 500),
+ range(500, 501), range(501, Long.MAX_VALUE));
+ assertThat(consolidated.getRangeMap().asMapOfRanges())
+ .describedAs("Consolidated RangeMap has different key set (ranges) from both topologies because the range layouts are different")
+ .hasSize(topology1.getRangeMap().asMapOfRanges().size() + topology2.getRangeMap().asMapOfRanges().size() - 1)
+ .containsOnlyKeys(expectedConsolidatedRanges);
+ // the instances per range in the consolidated should contain the instances of the same range from both topologies
+ expectedConsolidatedRanges.forEach(range -> {
+ List<RingInstance> instances = consolidated.getRangeMap().asMapOfRanges().get(range);
+ if (instances.isEmpty())
+ {
+ return;
+ }
+ List<RingInstance> instancesOfCluster1 = getSoleValue(topology1.getRangeMap().subRangeMap(range).asMapOfRanges());
+ List<RingInstance> instancesOfCluster2 = getSoleValue(topology2.getRangeMap().subRangeMap(range).asMapOfRanges());
+ assertThat(instances)
+ .hasSize(instancesOfCluster1.size() + instancesOfCluster2.size())
+ .containsAll(instancesOfCluster1)
+ .containsAll(instancesOfCluster2);
+ });
+ }
+
private TokenRangeMapping<RingInstance> createTestMapping(int instanceCount)
+ {
+ return createTestMapping(instanceCount, Partitioner.Murmur3Partitioner);
+ }
+
+ private TokenRangeMapping<RingInstance> createTestMapping(int instanceCount, Partitioner partitioner)
+ {
+ return createTestMapping(0L, instanceCount, partitioner, null);
+ }
+
+ private TokenRangeMapping<RingInstance> createTestMapping(long startToken, int instanceCount, Partitioner partitioner, String clusterId)
{
return TokenRangeMapping.create(
- () -> TokenRangeMappingUtils.mockSimpleTokenRangeReplicasResponse(instanceCount, 3),
- () -> Partitioner.Murmur3Partitioner,
- () -> rf,
- RingInstance::new);
+ () -> TokenRangeMappingUtils.mockSimpleTokenRangeReplicasResponse(startToken, instanceCount, 3),
+ () -> partitioner,
+ metadata -> new RingInstance(metadata, clusterId));
+ }
+
+ private Range<BigInteger> range(long start, long end)
+ {
+ return Range.openClosed(BigInteger.valueOf(start), BigInteger.valueOf(end));
+ }
+
+ private List<RingInstance> getSoleValue(Map<?, List<RingInstance>> map)
+ {
+ if (map.isEmpty())
+ {
+ return Collections.emptyList();
+ }
+
+ assertThat(map).hasSize(1);
+ return map.values().iterator().next();
}
}
diff --git a/cassandra-analytics-integration-framework/build.gradle b/cassandra-analytics-integration-framework/build.gradle
index aca0b9f..49e29a2 100644
--- a/cassandra-analytics-integration-framework/build.gradle
+++ b/cassandra-analytics-integration-framework/build.gradle
@@ -65,7 +65,7 @@ dependencies {
api("org.junit.vintage:junit-vintage-engine:${junitVersion}")
api('org.mockito:mockito-inline:4.10.0')
- api("org.assertj:assertj-core:3.24.2")
+ api("org.assertj:assertj-core:${assertjCoreVersion}")
api('com.datastax.cassandra:cassandra-driver-core:3.11.3')
implementation('com.google.inject:guice:4.2.2')
@@ -83,7 +83,7 @@ dependencies {
testImplementation(platform("org.junit:junit-bom:${project.junitVersion}"))
testImplementation('org.junit.jupiter:junit-jupiter')
- testImplementation("org.assertj:assertj-core:3.24.2")
+ testImplementation("org.assertj:assertj-core:${assertjCoreVersion}")
// Exposed to the test project
api("io.vertx:vertx-junit5:${project.vertxVersion}")
diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlCollection.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlCollection.java
index d005cc1..4f7f9b4 100644
--- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlCollection.java
+++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlCollection.java
@@ -51,11 +51,10 @@ public abstract class CqlCollection extends CqlType implements CqlField.CqlColle
this(Arrays.asList(types));
}
- @SuppressWarnings("ConfusingArgumentToVarargsMethod")
CqlCollection(List<CqlField.CqlType> types)
{
this.types = new ArrayList<>(types);
- this.hashCode = Objects.hash((CqlField.CqlType[]) types.toArray(new CqlField.CqlType[0]));
+ this.hashCode = Objects.hash(types.toArray());
}
@Override
diff --git a/gradle.properties b/gradle.properties
index 42413e4..aa7c785 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -23,6 +23,7 @@ description=Apache Cassandra Spark Analytics
analyticsJDKLevel=11
intellijVersion=9.0.4
junitVersion=5.10.2
+assertjCoreVersion=3.24.2
quickTheoriesVersion=0.26
mockitoVersion=3.12.4
jnaVersion=5.9.0
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org