This is an automated email from the ASF dual-hosted git repository.
frankgh pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new b31a451 CASSANDRA-19727: Bulk writer fails validation stage when
writing to a cluster using RandomPartitioner (#61)
b31a451 is described below
commit b31a451873eb411788a6d94c3cacf881f5c3cb86
Author: Francisco Guerrero <[email protected]>
AuthorDate: Thu Jun 27 15:51:15 2024 -0700
CASSANDRA-19727: Bulk writer fails validation stage when writing to a
cluster using RandomPartitioner (#61)
Patch by Francisco Guerrero; Reviewed by Yifan Cai for CASSANDRA-19727
---
CHANGES.txt | 1 +
.../spark/bulkwriter/SortedSSTableWriter.java | 12 +++++-
.../distributed/impl/CassandraCluster.java | 26 ++++++++++---
.../testing/ClusterBuilderConfiguration.java | 13 +++++++
.../cassandra/testing/TestTokenSupplier.java | 45 +++++++++++++++++-----
...Test.java => CassandraAnalyticsSimpleTest.java} | 5 ++-
.../cassandra/analytics/RandomPartitionerTest.java | 40 +++++++++++++++++++
7 files changed, 125 insertions(+), 17 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 2e654ef..614fb8e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.0.0
+ * Bulk writer fails validation stage when writing to a cluster using
RandomPartitioner (CASSANDRA-19727)
* Invalid mapping when timestamp is used as a partition key during bulk
writes (CASSANDRA-19716)
* NullPointerException when reading static column with null values
(CASSANDRA-19626)
* Integrate with the latest sidecar client (CASSANDRA-19616)
diff --git
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java
index 839beef..ea2cfc1 100644
---
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java
+++
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java
@@ -42,6 +42,7 @@ import org.apache.cassandra.spark.common.Digest;
import org.apache.cassandra.spark.common.SSTables;
import org.apache.cassandra.spark.data.DataLayer;
import org.apache.cassandra.spark.data.LocalDataLayer;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
import org.apache.cassandra.spark.reader.Rid;
import org.apache.cassandra.spark.reader.StreamScanner;
import org.apache.cassandra.spark.utils.DigestAlgorithm;
@@ -181,9 +182,18 @@ public class SortedSSTableWriter
CassandraVersion version =
CassandraBridgeFactory.getCassandraVersion(writerContext.cluster().getLowestCassandraVersion());
String keyspace =
writerContext.job().qualifiedTableName().keyspace();
String schema =
writerContext.schema().getTableSchema().createStatement;
+ Partitioner partitioner = writerContext.cluster().getPartitioner();
Set<String> udtStatements =
writerContext.schema().getUserDefinedTypeStatements();
String directory = getOutDir().toString();
- DataLayer layer = new LocalDataLayer(version, keyspace, schema,
udtStatements, directory);
+ DataLayer layer = new LocalDataLayer(version,
+ partitioner,
+ keyspace,
+ schema,
+ udtStatements,
+ Collections.emptyList() /*
requestedFeatures */,
+ false /*
useSSTableInputStream */,
+ null /* statsClass */,
+ directory);
try (StreamScanner<Rid> scanner =
layer.openCompactionScanner(partitionId, Collections.emptyList(), null))
{
while (scanner.hasNext())
diff --git
a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java
b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java
index 09c5e47..dcb6364 100644
---
a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java
+++
b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java
@@ -43,6 +43,7 @@ import org.apache.cassandra.distributed.api.IMessageSink;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.shared.Versions;
+import org.apache.cassandra.sidecar.adapters.base.Partitioner;
import org.apache.cassandra.testing.ClusterBuilderConfiguration;
import org.apache.cassandra.testing.IClusterExtension;
import org.apache.cassandra.testing.TestTokenSupplier;
@@ -92,16 +93,31 @@ public class CassandraCluster<I extends IInstance>
implements IClusterExtension<
int originalNodeCount = nodesPerDc * dcCount;
int finalNodeCount = dcCount * (nodesPerDc + newNodesPerDc);
- TokenSupplier tokenSupplier =
TestTokenSupplier.evenlyDistributedTokens(nodesPerDc, newNodesPerDc, dcCount,
1);
-
UpgradeableCluster.Builder clusterBuilder =
UpgradeableCluster.build(originalNodeCount);
clusterBuilder.withVersion(requestedVersion)
.withDynamicPortAllocation(configuration.dynamicPortAllocation) // to allow
parallel test runs
.withSharedClasses(EXTRA.or(clusterBuilder.getSharedClasses()))
.withDCs(dcCount)
- .withDataDirCount(configuration.numDataDirsPerInstance)
- .withConfig(config ->
configuration.features.forEach(config::with))
- .withTokenSupplier(tokenSupplier);
+ .withDataDirCount(configuration.numDataDirsPerInstance);
+
+ TokenSupplier tokenSupplier;
+ Consumer<IInstanceConfig> instanceConfigUpdater;
+ if (configuration.partitioner != null)
+ {
+ tokenSupplier =
TestTokenSupplier.evenlyDistributedTokens(Partitioner.fromClassName(configuration.partitioner),
+
nodesPerDc, newNodesPerDc, dcCount, 1);
+ instanceConfigUpdater = instanceConfig -> {
+ instanceConfig.set("partitioner", configuration.partitioner);
+ configuration.features.forEach(instanceConfig::with);
+ };
+ }
+ else
+ {
+ tokenSupplier =
TestTokenSupplier.evenlyDistributedTokens(nodesPerDc, newNodesPerDc, dcCount,
1);
+ instanceConfigUpdater = config ->
configuration.features.forEach(config::with);
+ }
+ clusterBuilder.withTokenSupplier(tokenSupplier)
+ .withConfig(instanceConfigUpdater);
if (dcCount > 1)
{
diff --git
a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/ClusterBuilderConfiguration.java
b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/ClusterBuilderConfiguration.java
index 91ff516..8759e3c 100644
---
a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/ClusterBuilderConfiguration.java
+++
b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/ClusterBuilderConfiguration.java
@@ -38,6 +38,7 @@ public class ClusterBuilderConfiguration
public boolean dynamicPortAllocation = true;
public final EnumSet<Feature> features = EnumSet.of(Feature.GOSSIP,
Feature.JMX, Feature.NATIVE_PROTOCOL);
public BiConsumer<ClassLoader, Integer> instanceInitializer = null;
+ public String partitioner;
/**
* Adds a features to the list of default features.
@@ -136,4 +137,16 @@ public class ClusterBuilderConfiguration
this.dynamicPortAllocation = dynamicPortAllocation;
return this;
}
+
+ /**
+ * Sets the {@code partitioner} and returns a reference to this Builder
enabling method chaining.
+ *
+ * @param partitioner the {@code partitioner} to set
+ * @return a reference to this Builder
+ */
+ public ClusterBuilderConfiguration partitioner(String partitioner)
+ {
+ this.partitioner = partitioner;
+ return this;
+ }
}
diff --git
a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestTokenSupplier.java
b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestTokenSupplier.java
index 28e8e4f..4268904 100644
---
a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestTokenSupplier.java
+++
b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestTokenSupplier.java
@@ -20,9 +20,11 @@ package org.apache.cassandra.testing;
import java.math.BigInteger;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.sidecar.adapters.base.Partitioner;
/**
* Static factory holder that provides a token supplier
@@ -38,7 +40,8 @@ public class TestTokenSupplier
/**
* Tokens are allocation used in tests to simulate token allocation nodes
for an approx even distribution
* in a multiDC environment with neighboring nodes different DCs allocated
adjacent tokens.
- * Allocations for new nodes are interleaved among existing tokens.
+ * Allocations for new nodes are interleaved among existing tokens. Usings
the Murmur3 Partitioner to allocate
+ * tokens.
*
* @param numNodesPerDC number of nodes from a single DC
* @param newNodesPerDC number of additional nodes for a DC
@@ -47,14 +50,38 @@ public class TestTokenSupplier
* @return The token supplier that vends the tokens
*/
public static TokenSupplier evenlyDistributedTokens(int numNodesPerDC, int
newNodesPerDC, int numDcs, int numTokensPerNode)
+ {
+ return evenlyDistributedTokens(Partitioner.Murmur3, numNodesPerDC,
newNodesPerDC, numDcs, numTokensPerNode);
+ }
+
+ /**
+ * Tokens are allocation used in tests to simulate token allocation nodes
for an approx even distribution
+ * in a multiDC environment with neighboring nodes different DCs allocated
adjacent tokens.
+ * Allocations for new nodes are interleaved among existing tokens.
+ *
+ * @param partitioner the partitioner to use for token allocation
+ * @param numNodesPerDC number of nodes from a single DC
+ * @param newNodesPerDC number of additional nodes for a DC
+ * @param numDcs no. of datacenters
+ * @param numTokensPerNode no. tokens allocated to each node (this is
always 1 if there are no vnodes)
+ * @return The token supplier that vends the tokens
+ */
+ public static TokenSupplier evenlyDistributedTokens(Partitioner
partitioner,
+ int numNodesPerDC,
+ int newNodesPerDC,
+ int numDcs,
+ int numTokensPerNode)
{
// Use token count using initial node count to first assign tokens to
nodes
long totalTokens = (long) (numNodesPerDC) * numDcs * numTokensPerNode;
// Similar to Cassandra TokenSupplier, the increment is doubled to
account for all tokens from MIN - MAX.
// For multi-DC, since neighboring nodes from different DCs have
consecutive tokens, the increment is
// broadened by a factor of numDcs.
- BigInteger increment = BigInteger.valueOf(((Long.MAX_VALUE /
(totalTokens + 2)) * 2 * numDcs));
- List<String>[] tokens = allocateExistingNodeTokens(numNodesPerDC,
+ BigInteger increment =
partitioner.maxToken.subtract(partitioner.minToken.add(BigInteger.ONE))
+
.multiply(BigInteger.valueOf(numDcs))
+
.divide(BigInteger.valueOf(totalTokens + 2));
+ List<String>[] tokens = allocateExistingNodeTokens(partitioner,
+ numNodesPerDC,
newNodesPerDC,
numDcs,
numTokensPerNode,
@@ -62,7 +89,7 @@ public class TestTokenSupplier
// Initial value of the first new node
- BigInteger value = new BigInteger(tokens[(numDcs - 1)].get(0));
+ BigInteger value = new BigInteger(tokens[(numDcs - 1)].get(0));
BigInteger subIncrement = increment.divide(BigInteger.valueOf(2));
int nodeId = (int) totalTokens + 1;
@@ -84,19 +111,17 @@ public class TestTokenSupplier
return (nodeIdx) -> tokens[nodeIdx - 1];
}
- private static List<String>[] allocateExistingNodeTokens(int numNodesPerDC,
+ private static List<String>[] allocateExistingNodeTokens(Partitioner
partitioner,
+ int numNodesPerDC,
int newNodesPerDC,
int numDcs,
int
numTokensPerNode,
BigInteger
increment)
{
List<String>[] tokens = new List[(numNodesPerDC + newNodesPerDC) *
numDcs];
- for (int i = 0; i < ((numNodesPerDC + newNodesPerDC) * numDcs); ++i)
- {
- tokens[i] = new ArrayList(numTokensPerNode);
- }
+ Arrays.setAll(tokens, ignored -> new ArrayList<>(numTokensPerNode));
- BigInteger value = BigInteger.valueOf(Long.MIN_VALUE + 1);
+ BigInteger value = partitioner.minToken.add(BigInteger.ONE);
int nodeId = 1;
for (int i = 0; i < numTokensPerNode; ++i)
diff --git
a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkBulkWriterSimpleTest.java
b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/CassandraAnalyticsSimpleTest.java
similarity index 96%
rename from
cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkBulkWriterSimpleTest.java
rename to
cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/CassandraAnalyticsSimpleTest.java
index 3c7f156..fa94401 100644
---
a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkBulkWriterSimpleTest.java
+++
b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/CassandraAnalyticsSimpleTest.java
@@ -40,7 +40,10 @@ import static org.apache.cassandra.testing.TestUtils.DC1_RF3;
import static org.apache.cassandra.testing.TestUtils.ROW_COUNT;
import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
-class SparkBulkWriterSimpleTest extends SharedClusterSparkIntegrationTestBase
+/**
+ * A simple test that runs a sample read/write Cassandra Analytics job.
+ */
+class CassandraAnalyticsSimpleTest extends
SharedClusterSparkIntegrationTestBase
{
static final QualifiedName QUALIFIED_NAME = new
QualifiedName(TEST_KEYSPACE, "test");
diff --git
a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/RandomPartitionerTest.java
b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/RandomPartitionerTest.java
new file mode 100644
index 0000000..6f7dfa0
--- /dev/null
+++
b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/RandomPartitionerTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cassandra.analytics;
+
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
+
+/**
+ * Runs test from {@link CassandraAnalyticsSimpleTest} with the {@code
RandomPartitioner}.
+ */
+class RandomPartitionerTest extends CassandraAnalyticsSimpleTest
+{
+ /**
+ * Inherit the configuration for the cluster from {@link
CassandraAnalyticsSimpleTest} and set the
+ * partitioner to be the RandomPartitioner.
+ *
+ * @return the updated configuration for the cluster
+ */
+ @Override
+ protected ClusterBuilderConfiguration testClusterConfiguration()
+ {
+ return
super.testClusterConfiguration().partitioner("org.apache.cassandra.dht.RandomPartitioner");
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
