From 6064e02e7c896ae8f15a6cd4f0e9ad3559df1de0 Mon Sep 17 00:00:00 2001 From: Saranya Krishnakumar Date: Tue, 7 Jul 2026 16:13:55 -0700 Subject: [PATCH 01/11] CASSANALYTICS-160: Determine whether mutation tracking is enabled for keyspace for bulk writes (#214) patch by Saranya Krishnakumar; reviewed by Shailaja Koppu, Yifan Cai for CASSANALYTICS-160 --- CHANGES.txt | 1 + .../cassandra/spark/utils/CqlUtils.java | 32 ++++ .../bulkwriter/CassandraClusterInfo.java | 11 ++ .../CassandraDirectDataTransportContext.java | 31 +++- .../spark/bulkwriter/ClusterInfo.java | 10 ++ .../TrackedDirectStreamSession.java | 70 ++++++++ .../bulkwriter/CassandraClusterInfoTest.java | 81 ++++++++++ ...ssandraDirectDataTransportContextTest.java | 151 ++++++++++++++++++ .../bulkwriter/MockBulkWriterContext.java | 12 ++ .../cassandra/spark/utils/CqlUtilsTest.java | 37 +++++ .../cassandra/bridge/CassandraBridge.java | 13 ++ 11 files changed, 441 insertions(+), 8 deletions(-) create mode 100644 cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TrackedDirectStreamSession.java create mode 100644 cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContextTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 974b9ceaa..19a658dc1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Determine whether mutation tracking is enabled for keyspace for bulk writes (CASSANALYTICS-160) * Upgrade sidecar version to 0.4.0 * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175) * Regenerate bloom filters for CQLSSTableWriter (CASSANALYTICS-167) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java index 9d989be51..8511dbaa1 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java @@ -52,6 +52,9 @@ public final class CqlUtils "max_index_interval" ); private static final Pattern REPLICATION_FACTOR_PATTERN = Pattern.compile("WITH REPLICATION = (\\{[^\\}]*\\})"); + private static final String TRACKED_REPLICATION_TYPE = "tracked"; + private static final Pattern REPLICATION_TYPE_PATTERN = Pattern.compile("replication_type\\s*=\\s*'(\\w+)'", + Pattern.CASE_INSENSITIVE); // Initialize a mapper allowing single quotes to process the RF string from the CREATE KEYSPACE statement private static final ObjectMapper MAPPER = new ObjectMapper().configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); private static final Pattern ESCAPED_WHITESPACE_PATTERN = Pattern.compile("(\\\\r|\\\\n|\\\\r\\n)+"); @@ -296,4 +299,33 @@ public static boolean isTimeRangeFilterSupported(String compactionStrategy) { return compactionStrategy == null || compactionStrategy.endsWith("TimeWindowCompactionStrategy"); } + + /** + * Extracts replication type from create schema statement + * + * @param schemaStr full cluster schema string as returned by Sidecar + * @param keyspace name of the keyspace to check + * @return {@code true} if keyspace is tracked {@code false} otherwise + */ + public static String extractReplicationType(@NotNull String schemaStr, @NotNull String keyspace) + { + String createKeyspaceSchema = extractKeyspaceSchema(schemaStr, keyspace); + Matcher matcher = REPLICATION_TYPE_PATTERN.matcher(createKeyspaceSchema); + if (matcher.find()) + { + return matcher.group(1); + } + return null; + } + + /** + * Returns {@code true} if {@code replication_type = 'tracked'} in create statement otherwise {@code false} + * + * @param replicationType replication type extracted from create statement + * @return {@code true} if replication type is tracked {@code false} otherwise + */ + public static boolean isTracked(String replicationType) + { + return TRACKED_REPLICATION_TYPE.equalsIgnoreCase(replicationType); + } } 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 a9c120871..53ac97974 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 @@ -356,6 +356,17 @@ public ReplicationFactor replicationFactor() } } + @Override + public String getReplicationType() + { + String keyspaceSchema = getKeyspaceSchema(true); + if (keyspaceSchema == null) + { + throw new RuntimeException("Could not retrieve keyspace schema information for keyspace " + conf.keyspace); + } + return CqlUtils.extractReplicationType(keyspaceSchema, conf.keyspace); + } + @Override public TokenRangeMapping getTokenRangeMapping(boolean cached) { diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java index bed36ac4b..fb82483a7 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java @@ -46,13 +46,24 @@ public CassandraDirectDataTransportContext(@NotNull BulkWriterContext bulkWriter } @Override - public DirectStreamSession createStreamSession(BulkWriterContext writerContext, - String sessionId, - SortedSSTableWriter sstableWriter, - Range range, - ReplicaAwareFailureHandler failureHandler, - ExecutorService executorService) + public StreamSession createStreamSession( + BulkWriterContext writerContext, + String sessionId, + SortedSSTableWriter sstableWriter, + Range range, + ReplicaAwareFailureHandler failureHandler, + ExecutorService executorService) { + if (bridge().isTracked(clusterInfo.getReplicationType())) + { + return new TrackedDirectStreamSession(writerContext, + sstableWriter, + this, + sessionId, + range, + failureHandler, + executorService); + } return new DirectStreamSession(writerContext, sstableWriter, this, @@ -71,7 +82,11 @@ public DirectDataTransferApi dataTransferApi() // only invoke in constructor protected DirectDataTransferApi createDirectDataTransferApi() { - CassandraBridge bridge = CassandraBridgeFactory.get(clusterInfo.getLowestCassandraVersion()); - return new SidecarDataTransferApi(clusterInfo.getCassandraContext(), bridge, jobInfo); + return new SidecarDataTransferApi(clusterInfo.getCassandraContext(), bridge(), jobInfo); + } + + private CassandraBridge bridge() + { + return CassandraBridgeFactory.get(clusterInfo.getLowestCassandraVersion()); } } 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 64b654449..e4d8a0a0b 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 @@ -82,6 +82,16 @@ public interface ClusterInfo extends StartupValidatable */ ReplicationFactor replicationFactor(); + /** + * @return {@code replication_type} of the enclosing keyspace (e.g. {@code "tracked"}, {@code "untracked"}), + * or {@code null} if replication_type is absent + */ + @Nullable + default String getReplicationType() + { + return null; + } + CassandraContext getCassandraContext(); /** diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TrackedDirectStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TrackedDirectStreamSession.java new file mode 100644 index 000000000..166ef2387 --- /dev/null +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TrackedDirectStreamSession.java @@ -0,0 +1,70 @@ +/* + * 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.util.Set; +import java.util.concurrent.ExecutorService; + +import com.google.common.collect.Range; + +import org.apache.cassandra.bridge.SSTableDescriptor; +import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler; + +/** + * Stream session for bulk writes to keyspaces with mutation tracking enabled. + * + *

+ * Tracked stream session uploads and triggers import on the coordinator node only. Cassandra's coordinated + * transfer then propagates the data to all other replicas, avoiding the duplicate-row updates that would occur if each + * replica independently streamed the data to its peers. + *

+ */ +public class TrackedDirectStreamSession extends StreamSession +{ + public TrackedDirectStreamSession(BulkWriterContext writerContext, + SortedSSTableWriter sstableWriter, + TransportContext.DirectDataBulkWriterContext transportContext, + String sessionID, + Range tokenRange, + ReplicaAwareFailureHandler failureHandler, + ExecutorService executorService) + { + super(writerContext, sstableWriter, transportContext, sessionID, tokenRange, failureHandler, executorService); + } + + @Override + protected void onSSTablesProduced(Set sstables) + { + throw new UnsupportedOperationException("TrackedDirectStreamSession is not yet implemented"); + } + + @Override + protected StreamResult doFinalizeStream() + { + throw new UnsupportedOperationException("TrackedDirectStreamSession is not yet implemented"); + } + + @Override + protected void sendRemainingSSTables() + { + throw new UnsupportedOperationException("TrackedDirectStreamSession is not yet implemented"); + } +} 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 index 5ed4c5221..763082a36 100644 --- 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 @@ -30,8 +30,10 @@ 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 org.apache.cassandra.spark.utils.CqlUtils; import static org.apache.cassandra.spark.TestUtils.range; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; @@ -75,6 +77,50 @@ public static CassandraClusterInfo mockClusterInfoForTimeSkewTest(int allowanceM return new MockClusterInfoForTimeSkew(allowanceMinutes, remoteNow); } + @Test + void testGetTrackedReplicationType() + { + String schema = "CREATE KEYSPACE mykeyspace " + + "WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': '3'}" + + " AND durable_writes = true" + + " AND replication_type = 'tracked';"; + CassandraClusterInfo ci = mockClusterInfoWithSchema("mykeyspace", schema); + assertThat(ci.getReplicationType()) + .describedAs("Keyspace with replication_type = 'tracked' should return 'tracked'") + .isEqualTo("tracked"); + } + + @Test + void testGetUntrackedReplicationType() + { + String schema = "CREATE KEYSPACE k " + + "WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': '3'}" + + " AND durable_writes = true" + + " AND replication_type = 'untracked';"; + CassandraClusterInfo ci = mockClusterInfoWithSchema("k", schema); + assertThat(ci.getReplicationType()) + .describedAs("Keyspace with replication_type = 'untracked' should return 'untracked'") + .isEqualTo("untracked"); + } + + @Test + void testGetReplicationTypeReturnsNullWhenPropertyAbsent() + { + // Schema without replication_type — pre-mutation-tracking clusters + String schema = "CREATE KEYSPACE k " + + "WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': '3'}" + + " AND durable_writes = true;"; + CassandraClusterInfo ci = mockClusterInfoWithSchema("k", schema); + assertThat(ci.getReplicationType()) + .describedAs("Keyspace without replication_type property should return null") + .isNull(); + } + + private static CassandraClusterInfo mockClusterInfoWithSchema(String keyspace, String schemaStr) + { + return new MockClusterInfoForSchema(keyspace, schemaStr); + } + private static class MockClusterInfoForTimeSkew extends CassandraClusterInfo { private CassandraContext cassandraContext; @@ -107,4 +153,39 @@ private void mockCassandraContext(int allowanceMinutes, Instant remoteNow) when(cassandraContext.sidecarPort()).thenReturn(9043); } } + + /** + * Minimal ClusterInfo stub that returns a fixed keyspace schema string, used to test + * {@link CassandraClusterInfo#getReplicationType()}. + */ + private static class MockClusterInfoForSchema extends CassandraClusterInfo + { + private final String keyspaceName; + private final String schemaStr; + + MockClusterInfoForSchema(String keyspace, String schemaStr) + { + super((BulkSparkConf) null); + this.keyspaceName = keyspace; + this.schemaStr = schemaStr; + } + + @Override + protected CassandraContext buildCassandraContext() + { + return mock(CassandraContext.class, RETURNS_DEEP_STUBS); + } + + @Override + public String getKeyspaceSchema(boolean cached) + { + return schemaStr; + } + + @Override + public String getReplicationType() + { + return CqlUtils.extractReplicationType(schemaStr, keyspaceName); + } + } } diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContextTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContextTest.java new file mode 100644 index 000000000..0bd2170bb --- /dev/null +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContextTest.java @@ -0,0 +1,151 @@ +/* + * 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.nio.file.Path; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import com.google.common.collect.BoundType; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Range; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.apache.cassandra.spark.bulkwriter.token.MultiClusterReplicaAwareFailureHandler; +import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler; +import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping; +import org.apache.cassandra.spark.data.ReplicationFactor; +import org.apache.cassandra.spark.utils.XXHash32DigestAlgorithm; + +import static org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that {@link CassandraDirectDataTransportContext#createStreamSession}. + */ +public class CassandraDirectDataTransportContextTest +{ + @TempDir + private Path folder; + + private MockBulkWriterContext writerContext; + private MockTableWriter tableWriter; + private ExecutorService executor; + private Range range; + + @BeforeEach + public void setup() + { + ImmutableMap rfOptions = ImmutableMap.of("DC1", 3); + ReplicationFactor rf = new ReplicationFactor(NetworkTopologyStrategy, rfOptions); + TokenRangeMapping tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(0, rfOptions, 12); + writerContext = new MockBulkWriterContext(tokenRangeMapping); + writerContext.setReplicationFactor(rf); + tableWriter = new MockTableWriter(folder); + range = Range.range(BigInteger.valueOf(101L), BoundType.CLOSED, BigInteger.valueOf(199L), BoundType.CLOSED); + executor = Executors.newSingleThreadExecutor(); + } + + @AfterEach + public void tearDown() + { + executor.shutdownNow(); + } + + @Test + void createDirectStreamSessionForUntrackedKeyspace() + { + writerContext.setTrackedKeyspace(false); + CassandraDirectDataTransportContext transportContext = stubTransportContext(); + + StreamSession session = transportContext.createStreamSession( + writerContext, + "session-untracked", + new SortedSSTableWriter(tableWriter, folder, new XXHash32DigestAlgorithm(), 1), + range, + buildFailureHandler(), + executor); + + assertThat(session) + .describedAs("Untracked keyspace should produce a DirectStreamSession") + .isInstanceOf(DirectStreamSession.class); + } + + @Test + void createTrackedDirectStreamSessionForTrackedKeyspace() + { + writerContext.setTrackedKeyspace(true); + CassandraDirectDataTransportContext transportContext = stubTransportContext(); + + StreamSession session = transportContext.createStreamSession( + writerContext, + "session-tracked", + new SortedSSTableWriter(tableWriter, folder, new XXHash32DigestAlgorithm(), 1), + range, + buildFailureHandler(), + executor); + + assertThat(session) + .describedAs("Tracked keyspace should produce a TrackedDirectStreamSession") + .isInstanceOf(TrackedDirectStreamSession.class); + } + + @Test + void createDirectStreamSessionForAbsentReplicationType() + { + CassandraDirectDataTransportContext transportContext = stubTransportContext(); + + StreamSession session = transportContext.createStreamSession( + writerContext, + "session-null-type", + new SortedSSTableWriter(tableWriter, folder, new XXHash32DigestAlgorithm(), 1), + range, + buildFailureHandler(), + executor); + + assertThat(session) + .describedAs("Null replication type (pre-mutation-tracking) should produce a DirectStreamSession") + .isInstanceOf(DirectStreamSession.class); + } + + private CassandraDirectDataTransportContext stubTransportContext() + { + DirectDataTransferApi mockApi = + ((TransportContext.DirectDataBulkWriterContext) writerContext.transportContext()).dataTransferApi(); + + return new CassandraDirectDataTransportContext(writerContext) + { + @Override + protected DirectDataTransferApi createDirectDataTransferApi() + { + return mockApi; + } + }; + } + + private ReplicaAwareFailureHandler buildFailureHandler() + { + return new MultiClusterReplicaAwareFailureHandler<>(writerContext.cluster().getPartitioner()); + } +} 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 cb5564f0c..be05715c4 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 @@ -99,6 +99,7 @@ public interface CommitResultSupplier extends BiFunction, String, D private final UUID jobId; private boolean skipClean = false; + private String replicationType = null; public int refreshClusterInfoCallCount = 0; // CHECKSTYLE IGNORE: Public mutable field private final Map> uploads = new ConcurrentHashMap<>(); private final Map> commits = new ConcurrentHashMap<>(); @@ -204,6 +205,17 @@ public void setReplicationFactor(ReplicationFactor replicationFactor) this.replicationFactor = replicationFactor; } + @Override + public String getReplicationType() + { + return replicationType; + } + + public void setTrackedKeyspace(boolean tracked) + { + this.replicationType = tracked ? "tracked" : "untracked"; + } + @Override public CassandraContext getCassandraContext() { diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java index 5bb363d97..58fb5c960 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java @@ -833,6 +833,43 @@ public void testTimeRangeFilterSupported() assertThat(isTimeRangeFilterSupported("")).isFalse(); } + @Test + public void testExtractReplicationType() + { + String schemaStrTracked + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true AND replication_type = 'tracked';"; + assertThat(CqlUtils.extractReplicationType(schemaStrTracked, "k")).isEqualTo("tracked"); + String schemaStrTrackedCaseInsensitive + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true AND REPLICATION_TYPE = 'Tracked';"; + assertThat(CqlUtils.extractReplicationType(schemaStrTrackedCaseInsensitive, "k")).isEqualTo("Tracked"); + String schemaStrUntracked + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true AND replication_type = 'untracked';"; + assertThat(CqlUtils.extractReplicationType(schemaStrUntracked, "k")).isEqualTo("untracked"); + String schemaStrNoReplicationType + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true;"; + assertThat(CqlUtils.extractReplicationType(schemaStrNoReplicationType, "k")).isNull(); + } + + @Test + public void testIsTracked() + { + assertThat(CqlUtils.isTracked("tracked")).isTrue(); + assertThat(CqlUtils.isTracked("TRACKED")).isTrue(); + assertThat(CqlUtils.isTracked("Tracked")).isTrue(); + assertThat(CqlUtils.isTracked("untracked")).isFalse(); + assertThat(CqlUtils.isTracked("random")).isFalse(); + assertThat(CqlUtils.isTracked(null)).isFalse(); + assertThat(CqlUtils.isTracked("")).isFalse(); + } + private static String loadFullSchemaSample() throws IOException { Path fullSchemaSampleFile = ResourceUtils.writeResourceToPath(CqlUtilsTest.class.getClassLoader(), tempPath, "cql/fullSchema.cql"); diff --git a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java index 67c5bc7a0..d2d67ab92 100644 --- a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java +++ b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java @@ -62,6 +62,7 @@ import org.apache.cassandra.spark.sparksql.filters.SparkRangeFilter; import org.apache.cassandra.analytics.stats.Stats; import org.apache.cassandra.spark.sparksql.filters.SSTableTimeRangeFilter; +import org.apache.cassandra.spark.utils.CqlUtils; import org.apache.cassandra.spark.utils.TimeProvider; import org.apache.cassandra.util.CompressionUtil; import org.jetbrains.annotations.NotNull; @@ -167,6 +168,18 @@ public String maybeQuoteIdentifier(String identifier) return cassandraTypes().maybeQuoteIdentifier(identifier); } + /** + * @return {@code true} if mutation tracking is enabled given the replication type, {@code false} otherwise + */ + public boolean isTracked(String replicationType) + { + if (replicationType == null) + { + return false; + } + return CqlUtils.isTracked(replicationType); + } + // CQL Type Parsing public CqlField.CqlType readType(CqlField.CqlType.InternalType type, Input input) From dda5235bdb56063d76afbf5daaf3c3e1d716e1a7 Mon Sep 17 00:00:00 2001 From: mkhara Date: Fri, 28 Aug 2026 16:46:54 -0700 Subject: [PATCH 02/11] CASSANALYTICS-194: Parse / replication factor for witness-enabled keyspaces Cassandra accepts a replication factor of the form /, which witness replicas under mutation tracking reuse, so a witness-enabled keyspace declares 'datacenter1': '3/1' meaning three replicas of which one is a witness. CqlUtils.extractReplicationFactor() passed each datacenter value to Integer.parseInt, so a bulk read against any such keyspace failed during job setup with an uncaught NumberFormatException. Transient counts are now tracked per datacenter alongside the existing totals and exposed through getFullReplicationFactor(), getTransientReplicationFactor(), getFullReplicas(dc), getTransientReplicas(dc) and hasTransientReplicas(). getTotalReplicationFactor() keeps its existing meaning of all replicas including witnesses, so behaviour for untracked keyspaces is unchanged. Parsing applies the same constraints Cassandra enforces in locator.ReplicationFactor.validate. A new parseStrict factory reports an unparseable or empty replication map at parse time rather than dropping the datacenter and failing later with a misleading "DC not found in replication factor"; the lenient constructor is retained unchanged for CDC callers. The Kryo serializer and CassandraRing's hand-rolled JDK readObject/writeObject are updated so the new field survives serialization to Spark executors. CassandraRing gains a pinned serialVersionUID and a format-version byte, because the class signature did not change and an older stream would otherwise be silently misread rather than rejected. Prerequisite for CASSANALYTICS-164. patch by Mansi Khara; reviewed by TBD for CASSANALYTICS-194 --- CHANGES.txt | 1 + .../spark/data/ReplicationFactor.java | 281 ++++++++++++++- .../spark/data/partitioner/CassandraRing.java | 36 +- .../cassandra/spark/utils/CqlUtils.java | 15 +- .../spark/data/ReplicationFactorTests.java | 328 ++++++++++++++++++ .../data/partitioner/CassandraRingTests.java | 104 ++++++ .../cassandra/spark/utils/CqlUtilsTest.java | 78 +++++ .../spark/reader/SchemaBuilderTests.java | 16 + .../spark/reader/SchemaBuilderTests.java | 17 + .../cassandra/spark/reader/SchemaBuilder.java | 4 + 10 files changed, 870 insertions(+), 10 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 19a658dc1..469bbe82b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Parse / replication factor for witness-enabled keyspaces (CASSANALYTICS-194) * Determine whether mutation tracking is enabled for keyspace for bulk writes (CASSANALYTICS-160) * Upgrade sidecar version to 0.4.0 * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java index 0b9b6821d..ef40bcad8 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java @@ -20,6 +20,7 @@ package org.apache.cassandra.spark.data; import java.io.Serializable; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; @@ -50,6 +51,11 @@ * "replication_factor" : 1 * } * } + *

+ * Replica counts may also use the {@code /} form, e.g. {@code "DC1" : "3/1"}, meaning three + * replicas of which one is transient. Witness replicas under mutation tracking (CEP-45/CEP-46) reuse this form, so + * {@code "3/1"} describes two full replicas and one witness. {@link #getTotalReplicationFactor()} continues to + * report all three; use {@link #getFullReplicationFactor()} for the count that holds the full data set. */ public class ReplicationFactor implements Serializable { @@ -108,11 +114,44 @@ public static ReplicationFactor simpleStrategy(int rf) private final ReplicationStrategy replicationStrategy; @NotNull private final Map options; + /** + * Per-datacenter count of transient (witness) replicas, parsed from the {@code /} form. + * A datacenter absent from this map has no transient replicas. Always empty for untracked keyspaces using the + * plain {@code } form, which keeps behaviour identical for those keyspaces. + */ + @NotNull + private final Map transientOptions; + /** + * Lenient parse: a replication value that cannot be parsed is logged and its datacenter omitted. Retained for + * callers that tolerate a partial replication factor. + * + * @param options the raw replication map, including the {@code class} entry + */ public ReplicationFactor(@NotNull Map options) + { + this(options, false); + } + + /** + * Strict parse: a replication value that cannot be parsed raises {@link IllegalArgumentException} naming the + * offending datacenter, rather than silently omitting it. Prefer this when a partial replication factor would + * produce a misleading failure later. + * + * @param options the raw replication map, including the {@code class} entry + * @return the parsed replication factor + * @throws IllegalArgumentException when any replication value cannot be parsed + */ + public static ReplicationFactor parseStrict(@NotNull Map options) + { + return new ReplicationFactor(options, true); + } + + private ReplicationFactor(@NotNull Map options, boolean strict) { this.replicationStrategy = ReplicationFactor.ReplicationStrategy.getEnum(options.get("class")); this.options = new LinkedHashMap<>(options.size()); + this.transientOptions = new LinkedHashMap<>(); for (Map.Entry entry : options.entrySet()) { if ("class".equals(entry.getKey())) @@ -122,21 +161,56 @@ public ReplicationFactor(@NotNull Map options) try { - this.options.put(entry.getKey(), Integer.parseInt(entry.getValue())); + ReplicaCounts counts = ReplicaCounts.parse(entry.getValue()); + this.options.put(entry.getKey(), counts.allReplicas); + if (counts.transientReplicas > 0) + { + this.transientOptions.put(entry.getKey(), counts.transientReplicas); + } } - catch (NumberFormatException exception) + catch (IllegalArgumentException exception) { + if (strict) + { + throw new IllegalArgumentException(String.format("Could not parse replication option: %s = %s", + entry.getKey(), entry.getValue()), exception); + } LOGGER.warn("Could not parse replication option: {} = {}", entry.getKey(), entry.getValue()); } } + + // Mirrors the guard on the (strategy, options) constructor. A strategy other than LocalStrategy with no + // datacenter entries is not usable, and reporting it here keeps the failure at parse time rather than + // surfacing later as a misleading "DC not found in replication factor". Strict-only, so the lenient + // constructor's behaviour is unchanged for callers that tolerate a partial replication factor. + if (strict && replicationStrategy != ReplicationStrategy.LocalStrategy && this.options.isEmpty()) + { + throw new IllegalArgumentException("Could not find replication info in schema map: " + options); + } } public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map options) + { + this(replicationStrategy, options, Collections.emptyMap()); + } + + public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, + @NotNull Map options, + @NotNull Map transientOptions) + { + this(replicationStrategy, options, transientOptions, true); + } + + private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, + @NotNull Map options, + @NotNull Map transientOptions, + boolean validate) { this.replicationStrategy = replicationStrategy; this.options = new LinkedHashMap<>(options.size()); + this.transientOptions = new LinkedHashMap<>(transientOptions.size()); - if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) + if (validate && !replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) { throw new RuntimeException(String.format("Could not find replication info in schema map: %s.", options)); } @@ -149,8 +223,28 @@ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotN } this.options.put(entry.getKey(), entry.getValue()); } + + for (Map.Entry entry : transientOptions.entrySet()) + { + if (entry.getValue() == null || entry.getValue() == 0) + { + continue; + } + Integer allReplicas = this.options.get(entry.getKey()); + if (allReplicas == null) + { + throw new IllegalArgumentException(String.format( + "Transient replicas specified for %s but it has no replication factor", entry.getKey())); + } + ReplicaCounts.validate(entry.getKey(), allReplicas, entry.getValue()); + this.transientOptions.put(entry.getKey(), entry.getValue()); + } } + /** + * @return the total number of replicas across all datacenters, including transient (witness) replicas. + * Semantics are unchanged from before transient replica support was added. + */ public Integer getTotalReplicationFactor() { return options.values().stream() @@ -158,12 +252,76 @@ public Integer getTotalReplicationFactor() .sum(); } + /** + * @return the number of replicas across all datacenters that hold the full data set, i.e. the total + * replication factor minus transient (witness) replicas + */ + public Integer getFullReplicationFactor() + { + return getTotalReplicationFactor() - getTransientReplicationFactor(); + } + + /** + * @return the number of transient (witness) replicas across all datacenters, {@code 0} when none are configured + */ + public Integer getTransientReplicationFactor() + { + return transientOptions.values().stream() + .mapToInt(Integer::intValue) + .sum(); + } + + /** + * @return {@code true} if any datacenter is configured with transient (witness) replicas + */ + public boolean hasTransientReplicas() + { + return !transientOptions.isEmpty(); + } + + /** + * @param datacenter the datacenter to look up + * @return the number of transient (witness) replicas in {@code datacenter}, {@code 0} when none are configured + */ + public int getTransientReplicas(@NotNull String datacenter) + { + return transientOptions.getOrDefault(datacenter, 0); + } + + /** + * @param datacenter the datacenter to look up + * @return the number of replicas in {@code datacenter} holding the full data set + * @throws IllegalArgumentException when {@code datacenter} has no replication factor + */ + public int getFullReplicas(@NotNull String datacenter) + { + Integer allReplicas = options.get(datacenter); + if (allReplicas == null) + { + throw new IllegalArgumentException(String.format("Datacenter %s not found in replication factor %s", + datacenter, options.keySet())); + } + return allReplicas - getTransientReplicas(datacenter); + } + + /** + * @return per-datacenter total replica counts, including transient (witness) replicas + */ @NotNull public Map getOptions() { return options; } + /** + * @return per-datacenter transient (witness) replica counts. Datacenters without transient replicas are absent. + */ + @NotNull + public Map getTransientOptions() + { + return transientOptions; + } + @NotNull public ReplicationStrategy getReplicationStrategy() { @@ -188,13 +346,110 @@ public boolean equals(Object other) ReplicationFactor that = (ReplicationFactor) other; return this.replicationStrategy == that.replicationStrategy - && java.util.Objects.equals(this.options, that.options); + && java.util.Objects.equals(this.options, that.options) + && java.util.Objects.equals(this.transientOptions, that.transientOptions); } @Override public int hashCode() { - return Objects.hash(replicationStrategy, options); + return Objects.hash(replicationStrategy, options, transientOptions); + } + + /** + * {@link #serialVersionUID} is pinned, so an instance serialized before transient replica support was added + * deserializes with a {@code null} {@code transientOptions}. Normalise it to an empty map so the accessors do not + * NPE. Only reachable under driver/executor version skew, which is not a supported configuration. + * + * @return this instance, or a normalised copy when deserialized from the older form + */ + private Object readResolve() + { + if (transientOptions != null) + { + return this; + } + // Bypasses validation deliberately: the older lenient constructor permitted a non-LocalStrategy + // replication factor with no datacenter entries, so re-validating here would turn a legacy instance + // into a deserialization failure + return new ReplicationFactor(replicationStrategy, options, Collections.emptyMap(), false); + } + + /** + * Parsed form of a single datacenter's replication value. Cassandra accepts either {@code } or + * {@code /}, the latter reused by witness replicas under mutation tracking. See + * {@code org.apache.cassandra.locator.ReplicationFactor} in Cassandra. + */ + private static final class ReplicaCounts + { + private static final String TRANSIENT_SEPARATOR = "/"; + + private final int allReplicas; + private final int transientReplicas; + + private ReplicaCounts(int allReplicas, int transientReplicas) + { + this.allReplicas = allReplicas; + this.transientReplicas = transientReplicas; + } + + /** + * @param value the raw replication value, e.g. {@code "3"} or {@code "3/1"} + * @return the parsed replica counts + * @throws NumberFormatException when either component is not an integer + * @throws IllegalArgumentException when the value is malformed or the counts are inconsistent + */ + static ReplicaCounts parse(@NotNull String value) + { + String trimmed = value.trim(); + int separator = trimmed.indexOf(TRANSIENT_SEPARATOR); + if (separator < 0) + { + int allReplicas = Integer.parseInt(trimmed); + validate(null, allReplicas, 0); + return new ReplicaCounts(allReplicas, 0); + } + + if (trimmed.indexOf(TRANSIENT_SEPARATOR, separator + 1) >= 0) + { + throw new IllegalArgumentException(String.format( + "Replication factor format is or /, found '%s'", value)); + } + + int allReplicas = Integer.parseInt(trimmed.substring(0, separator).trim()); + int transientReplicas = Integer.parseInt(trimmed.substring(separator + 1).trim()); + validate(null, allReplicas, transientReplicas); + return new ReplicaCounts(allReplicas, transientReplicas); + } + + /** + * Mirrors the constraints Cassandra enforces in {@code ReplicationFactor.validate}: transient replicas must + * be non-negative and strictly fewer than the total, so at least one full replica always exists. + * + * @param datacenter datacenter name for the error message, may be {@code null} + * @param allReplicas total replicas + * @param transientReplicas transient (witness) replicas + */ + static void validate(String datacenter, int allReplicas, int transientReplicas) + { + String where = datacenter == null ? "" : String.format(" for datacenter %s", datacenter); + if (allReplicas < 0) + { + throw new IllegalArgumentException(String.format( + "Replication factor must be non-negative, found %d%s", allReplicas, where)); + } + if (transientReplicas < 0) + { + throw new IllegalArgumentException(String.format( + "Transient replicas must be non-negative, found %d%s", transientReplicas, where)); + } + if (transientReplicas > 0 && transientReplicas >= allReplicas) + { + throw new IllegalArgumentException(String.format( + "Transient replicas must be zero, or less than the total replication factor. For %d/%d%s", + allReplicas, transientReplicas, where)); + } + } } public static class Serializer extends com.esotericsoftware.kryo.Serializer @@ -209,6 +464,14 @@ public void write(Kryo kryo, Output out, ReplicationFactor replicationFactor) out.writeString(entry.getKey()); out.writeByte(entry.getValue()); } + // Transient (witness) replica counts, written after the totals so the common + // no-transient-replicas case costs a single zero byte + out.writeByte(replicationFactor.transientOptions.size()); + for (Map.Entry entry : replicationFactor.transientOptions.entrySet()) + { + out.writeString(entry.getKey()); + out.writeByte(entry.getValue()); + } } @Override @@ -221,7 +484,13 @@ public ReplicationFactor read(Kryo kryo, Input in, Class type { options.put(in.readString(), (int) in.readByte()); } - return new ReplicationFactor(strategy, options); + int numTransientOptions = in.readByte(); + Map transientOptions = new HashMap<>(numTransientOptions); + for (int option = 0; option < numTransientOptions; option++) + { + transientOptions.put(in.readString(), (int) in.readByte()); + } + return new ReplicationFactor(strategy, options, transientOptions); } } } diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java index 3d142f8df..752ea65f6 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java @@ -69,6 +69,20 @@ public class CassandraRing implements Serializable private static final Logger LOGGER = LoggerFactory.getLogger(CassandraRing.class); public static final Serializer SERIALIZER = new Serializer(); + /** + * Pinned so that the JDK serialization format change made when transient (witness) replica counts were added is + * detected. Without it the UID is computed from the class signature, which did not change - only the + * {@link #readObject}/{@link #writeObject} bodies did - so an older stream would be silently misread rather than + * rejected. + */ + private static final long serialVersionUID = 2026082800000000001L; + + /** + * Incremented whenever the hand-rolled JDK serialization format below changes. Version 1 added the + * per-datacenter transient (witness) replica counts after the replication options. + */ + private static final byte SERIALIZATION_FORMAT_VERSION = 1; + private Partitioner partitioner; private String keyspace; private ReplicationFactor replicationFactor; @@ -260,6 +274,12 @@ public int hashCode() private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { LOGGER.debug("Falling back to JDK deserialization"); + byte formatVersion = in.readByte(); + if (formatVersion != SERIALIZATION_FORMAT_VERSION) + { + throw new IOException(String.format("Unsupported CassandraRing serialization format version %d, expected %d", + formatVersion, SERIALIZATION_FORMAT_VERSION)); + } this.partitioner = in.readByte() == 0 ? Partitioner.RandomPartitioner : Partitioner.Murmur3Partitioner; this.keyspace = in.readUTF(); @@ -270,7 +290,13 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE { options.put(in.readUTF(), (int) in.readByte()); } - this.replicationFactor = new ReplicationFactor(strategy, options); + int transientOptionCount = in.readByte(); + Map transientOptions = new HashMap<>(transientOptionCount); + for (int option = 0; option < transientOptionCount; option++) + { + transientOptions.put(in.readUTF(), (int) in.readByte()); + } + this.replicationFactor = new ReplicationFactor(strategy, options, transientOptions); int numInstances = in.readShort(); this.instances = new ArrayList<>(numInstances); @@ -284,6 +310,7 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFoundException { LOGGER.debug("Falling back to JDK serialization"); + out.writeByte(SERIALIZATION_FORMAT_VERSION); out.writeByte(this.partitioner == Partitioner.RandomPartitioner ? 0 : 1); out.writeUTF(this.keyspace); @@ -295,6 +322,13 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou out.writeUTF(option.getKey()); out.writeByte(option.getValue()); } + Map transientOptions = this.replicationFactor.getTransientOptions(); + out.writeByte(transientOptions.size()); + for (Map.Entry option : transientOptions.entrySet()) + { + out.writeUTF(option.getKey()); + out.writeByte(option.getValue()); + } out.writeShort(this.instances.size()); for (CassandraInstance instance : this.instances) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java index 8511dbaa1..b76814144 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java @@ -176,9 +176,18 @@ public static ReplicationFactor extractReplicationFactor(@NotNull String schemaS throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), exception); } - String className = map.remove("class"); - ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.getEnum(className); - return new ReplicationFactor(strategy, map.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, v -> Integer.parseInt(v.getValue())))); + // Values may use the / form for witness replicas, so delegate parsing to + // ReplicationFactor. parseStrict reports an unparseable value directly instead of dropping the + // datacenter, which would otherwise surface later as a confusing "DC not found" error. + try + { + return ReplicationFactor.parseStrict(map); + } + catch (IllegalArgumentException exception) + { + throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), + exception); + } } public static String extractTableSchema(@NotNull String schemaStr, @NotNull String keyspace, @NotNull String table) diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java index ef2eb8eeb..55883b4f7 100644 --- a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java +++ b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java @@ -19,8 +19,16 @@ package org.apache.cassandra.spark.data; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.lang.reflect.Field; import java.util.ArrayList; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; @@ -111,4 +119,324 @@ public void testEquality() assertThat(replicationFactor1).isEqualTo(replicationFactor2); assertThat(replicationFactor1.hashCode()).isEqualTo(replicationFactor2.hashCode()); } + + // Transient / witness replicas: the / form, reused by witness replicas under + // mutation tracking (CEP-45/CEP-46) + + @Test + public void testNoTransientReplicasByDefault() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3")); + assertThat(replicationFactor.hasTransientReplicas()).isFalse(); + assertThat(replicationFactor.getTransientOptions()).isEmpty(); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(0); + assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(3); + assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(0); + } + + @Test + public void testTransientReplicasSingleDatacenter() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThat(replicationFactor.hasTransientReplicas()).isTrue(); + // total keeps its original meaning: all replicas, witnesses included + assertThat(replicationFactor.getOptions().get("datacenter1")).isEqualTo(Integer.valueOf(3)); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(1); + assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(2); + assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1); + } + + @Test + public void testTransientReplicasMultipleDatacenters() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3/1")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(4); + assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(2); + } + + @Test + public void testMixedTransientAndFullDatacenters() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(5); + assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(replicationFactor.getTransientReplicas("datacenter2")).isEqualTo(0); + assertThat(replicationFactor.getFullReplicas("datacenter2")).isEqualTo(3); + // datacenter2 has no transient replicas, so it must be absent rather than mapped to zero + assertThat(replicationFactor.getTransientOptions()).containsOnlyKeys("datacenter1"); + } + + @Test + public void testTransientReplicasSimpleStrategy() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "SimpleStrategy", + "replication_factor", "3/1")); + assertThat(replicationFactor.getReplicationStrategy()) + .isEqualTo(ReplicationFactor.ReplicationStrategy.SimpleStrategy); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + } + + @Test + public void testTransientReplicasWithWhitespace() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", " 3 / 1 ")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + } + + @Test + public void testTransientEqualToTotalIsRejected() + { + // Cassandra requires at least one full replica, so 3/3 is invalid + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/3")); + assertThat(replicationFactor.getOptions()).doesNotContainKey("datacenter1"); + } + + @Test + public void testMalformedTransientValuesAreSkipped() + { + for (String malformed : new String[]{ "3/", "/1", "3/1/1", "3/x", "x/1", "3/-1", "" }) + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", malformed)); + assertThat(replicationFactor.getOptions()) + .as("malformed value '%s' should not produce a replication factor entry", malformed) + .doesNotContainKey("datacenter1"); + } + } + + @Test + public void testGetFullReplicasUnknownDatacenter() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThatThrownBy(() -> replicationFactor.getFullReplicas("nosuchdc")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testTransientOptionsForUnknownDatacenterRejected() + { + assertThatThrownBy(() -> new ReplicationFactor( + ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy, + ImmutableMap.of("datacenter1", 3), + ImmutableMap.of("datacenter2", 1))) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testEqualityConsidersTransientReplicas() + { + ReplicationFactor full = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3")); + ReplicationFactor withTransient = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThat(full).isNotEqualTo(withTransient); + assertThat(full.hashCode()).isNotEqualTo(withTransient.hashCode()); + } + + @Test + public void testNegativeReplicationFactorIsRejected() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "-3")); + assertThat(replicationFactor.getOptions()).doesNotContainKey("datacenter1"); + } + + @Test + public void testZeroReplicationFactorIsAllowed() + { + // RF 0 is legitimate for NetworkTopologyStrategy: the keyspace is simply not replicated to that datacenter + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3", + "datacenter2", "0")); + assertThat(replicationFactor.getOptions().get("datacenter2")).isEqualTo(Integer.valueOf(0)); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + } + + // parseStrict: same parsing, but an unparseable value raises instead of dropping the datacenter + + @Test + public void testParseStrictRaisesOnUnparseableValue() + { + assertThatThrownBy(() -> ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "xyz"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("datacenter1"); + } + + @Test + public void testParseStrictAcceptsTransientForm() + { + ReplicationFactor replicationFactor = ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + } + + @Test + public void testLenientConstructorStillDropsUnparseableValue() + { + // The lenient constructor is retained for callers that tolerate a partial replication factor + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3", + "datacenter2", "xyz")); + assertThat(replicationFactor.getOptions()).containsOnlyKeys("datacenter1"); + } + + @Test + public void testParseStrictRaisesWhenNoDatacenterEntries() + { + assertThatThrownBy(() -> ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Could not find replication info in schema map"); + } + + @Test + public void testParseStrictRaisesWhenEveryDatacenterIsUnparseable() + { + // Every entry dropped is the same situation as no entries at all, and must not yield an empty + // replication factor that fails later with a misleading message + assertThatThrownBy(() -> ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "xyz"))) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testParseStrictAllowsLocalStrategyWithNoEntries() + { + // LocalStrategy legitimately has no datacenter entries, e.g. the system_schema keyspace + ReplicationFactor replicationFactor = ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "org.apache.cassandra.locator.LocalStrategy")); + assertThat(replicationFactor.getReplicationStrategy()) + .isEqualTo(ReplicationFactor.ReplicationStrategy.LocalStrategy); + assertThat(replicationFactor.getOptions()).isEmpty(); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(0); + } + + @Test + public void testLenientConstructorAllowsNoDatacenterEntries() + { + // Unchanged lenient behaviour: no guard, so CDC callers are unaffected + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy")); + assertThat(replicationFactor.getOptions()).isEmpty(); + } + + // Serialization: a new field that silently fails to round-trip would surface as wrong + // replication data on Spark executors, so cover both paths with a non-zero transient count + + @Test + public void testKryoSerializationRoundTripWithTransientReplicas() throws Exception + { + ReplicationFactor original = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + + Kryo kryo = new Kryo(); + kryo.register(ReplicationFactor.class, new ReplicationFactor.Serializer()); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (Output out = new Output(bytes)) + { + kryo.writeObject(out, original); + } + ReplicationFactor deserialized; + try (Input in = new Input(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = kryo.readObject(in, ReplicationFactor.class); + } + + assertThat(deserialized).isEqualTo(original); + assertThat(deserialized.getTotalReplicationFactor()).isEqualTo(6); + assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5); + assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(deserialized.getTransientReplicas("datacenter2")).isEqualTo(0); + } + + @Test + public void testJdkSerializationRoundTripWithTransientReplicas() throws Exception { + ReplicationFactor original = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(original); + } + ReplicationFactor deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (ReplicationFactor) in.readObject(); + } + + assertThat(deserialized).isEqualTo(original); + assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5); + assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1); + } + + @Test + public void testReadResolveToleratesLegacyEmptyOptions() throws Exception + { + // An instance serialized before transient replica support deserializes with a null transientOptions, and + // the older lenient constructor allowed a non-LocalStrategy replication factor with no datacenter entries. + // readResolve must normalise it rather than turn it into a deserialization failure. + ReplicationFactor legacy = new ReplicationFactor(ImmutableMap.of("class", "NetworkTopologyStrategy")); + assertThat(legacy.getOptions()).isEmpty(); + + Field transientOptions = ReplicationFactor.class.getDeclaredField("transientOptions"); + transientOptions.setAccessible(true); + transientOptions.set(legacy, null); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(legacy); + } + ReplicationFactor deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (ReplicationFactor) in.readObject(); + } + + assertThat(deserialized.getOptions()).isEmpty(); + assertThat(deserialized.hasTransientReplicas()).isFalse(); + assertThat(deserialized.getTransientReplicationFactor()).isEqualTo(0); + } } diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java index c66290f0f..89cb14be3 100644 --- a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java +++ b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java @@ -19,6 +19,11 @@ package org.apache.cassandra.spark.data.partitioner; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.math.BigInteger; import java.util.Arrays; import java.util.Collection; @@ -34,6 +39,7 @@ import org.apache.cassandra.spark.data.ReplicationFactor; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; @SuppressWarnings("UnstableApiUsage") public class CassandraRingTests @@ -447,4 +453,102 @@ public void testNetworkStrategyRF22() Partitioner.Murmur3Partitioner.minToken(), Partitioner.Murmur3Partitioner.maxToken())); } + + @Test + public void testJdkSerializationPreservesTransientReplicas() throws Exception + { + // CassandraRing hand-rolls readObject/writeObject and rebuilds ReplicationFactor from strategy plus + // options, so a transient (witness) count could silently vanish on the way to a Spark executor + CassandraRing ring = new CassandraRing( + Partitioner.Murmur3Partitioner, + "test", + new ReplicationFactor(ImmutableMap.of("class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "DC1", "3/1", + "DC2", "3")), + Arrays.asList(new CassandraInstance("0", "local0-i1", "DC1"), + new CassandraInstance("100", "local0-i2", "DC1"), + new CassandraInstance("200", "local0-i3", "DC1"), + new CassandraInstance("1", "local1-i1", "DC2"), + new CassandraInstance("101", "local1-i2", "DC2"), + new CassandraInstance("201", "local1-i3", "DC2"))); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(ring); + } + CassandraRing deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (CassandraRing) in.readObject(); + } + + ReplicationFactor rf = deserialized.replicationFactor(); + assertThat(rf.getTransientReplicas("DC1")).isEqualTo(1); + assertThat(rf.getTransientReplicas("DC2")).isEqualTo(0); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(6); + assertThat(rf.getFullReplicationFactor()).isEqualTo(5); + assertThat(rf).isEqualTo(ring.replicationFactor()); + // Deliberately not asserting deserialized.equals(ring): CassandraRing#equals compares the derived + // replicas and tokenRangeMap fields, and does not hold across a JDK round trip even without transient + // replicas. Pre-existing behaviour, unrelated to replica types. + } + + @Test + public void testJdkDeserializationRejectsUnknownFormatVersion() throws Exception + { + // The hand-rolled format gained a transient-replica section, but the class signature did not change, so the + // computed serialVersionUID would still match an older stream. A leading format-version byte makes a stale + // stream fail loudly instead of being misread. + CassandraRing ring = new CassandraRing( + Partitioner.Murmur3Partitioner, + "test", + new ReplicationFactor(ImmutableMap.of("class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "DC1", "3")), + Arrays.asList(new CassandraInstance("0", "local0-i1", "DC1"), + new CassandraInstance("100", "local0-i2", "DC1"), + new CassandraInstance("200", "local0-i3", "DC1"))); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(ring); + } + + // Locate the format-version byte deterministically: the format is + // [version][partitioner] then writeUTF(keyspace), so it sits two bytes before the "test" UTF header + byte[] raw = bytes.toByteArray(); + byte[] keyspaceUtf = new byte[]{ 0, 4, 't', 'e', 's', 't' }; + int keyspaceIndex = indexOf(raw, keyspaceUtf); + assertThat(keyspaceIndex).as("keyspace UTF header should be locatable").isGreaterThan(1); + int versionIndex = keyspaceIndex - 2; + assertThat(raw[versionIndex]).as("format version byte").isEqualTo((byte) 1); + + // Simulate a stream written before the transient-replica section existed + raw[versionIndex] = 0; + + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(raw))) + { + assertThatThrownBy(in::readObject) + .isInstanceOf(IOException.class) + .hasMessageContaining("Unsupported CassandraRing serialization format version"); + } + } + + private static int indexOf(byte[] haystack, byte[] needle) + { + outer: + for (int i = 0; i <= haystack.length - needle.length; i++) + { + for (int j = 0; j < needle.length; j++) + { + if (haystack[i + j] != needle[j]) + { + continue outer; + } + } + return i; + } + return -1; + } } diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java index 58fb5c960..3a237a695 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java @@ -142,6 +142,84 @@ public void testExtractReplicationFactor(CassandraBridge bridge) assertThat(systemSchemaRf.getOptions()).isEqualTo(ImmutableMap.of()); } + @Test + public void testExtractReplicationFactorWithWitnessReplicas() + { + // Witness-enabled keyspace as created by Cassandra's WitnessAlwaysReadsFullReplicaTest on the + // cep-45-mutation-tracking branch: the / form plus replication_type = 'tracked' + String schema = "CREATE KEYSPACE witnessks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3/1', 'datacenter2': '3/1'} AND replication_type = 'tracked' " + + "AND durable_writes = true;\n"; + + ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "witnessks"); + assertThat(rf).isNotNull(); + assertThat(rf.getReplicationStrategy()).isEqualTo(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy); + // total keeps its original meaning: all replicas, witnesses included + assertThat(rf.getOptions()).isEqualTo(ImmutableMap.of("datacenter1", 3, "datacenter2", 3)); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(6); + assertThat(rf.getFullReplicationFactor()).isEqualTo(4); + assertThat(rf.getTransientReplicationFactor()).isEqualTo(2); + assertThat(rf.hasTransientReplicas()).isTrue(); + assertThat(rf.getFullReplicas("datacenter1")).isEqualTo(2); + assertThat(rf.getTransientReplicas("datacenter1")).isEqualTo(1); + + assertThat(CqlUtils.isTracked(CqlUtils.extractReplicationType(schema, "witnessks"))).isTrue(); + } + + @Test + public void testExtractReplicationFactorMixedWitnessAndFullDatacenters() + { + String schema = "CREATE KEYSPACE mixedks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3/1', 'datacenter2': '3'} AND replication_type = 'tracked';\n"; + + ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "mixedks"); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(6); + assertThat(rf.getFullReplicationFactor()).isEqualTo(5); + assertThat(rf.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(rf.getTransientReplicas("datacenter2")).isEqualTo(0); + } + + @Test + public void testExtractReplicationFactorUntrackedKeyspaceUnaffected() + { + String schema = "CREATE KEYSPACE plainks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3'} AND durable_writes = true;\n"; + + ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "plainks"); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(3); + assertThat(rf.getFullReplicationFactor()).isEqualTo(3); + assertThat(rf.hasTransientReplicas()).isFalse(); + assertThat(rf.getTransientOptions()).isEmpty(); + } + + @Test + public void testExtractReplicationFactorFailsLoudlyOnUnparseableValue() + { + // An unparseable value must be reported here rather than silently dropping the datacenter, which + // would surface later as a confusing "DC not found in replication factor" error + for (String malformed : new String[]{ "xyz", "3/", "3/1/1", "3/x", "3/3" }) + { + String schema = "CREATE KEYSPACE badks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '" + malformed + "'} AND durable_writes = true;\n"; + assertThatThrownBy(() -> CqlUtils.extractReplicationFactor(schema, "badks")) + .as("malformed replication value '%s' should raise", malformed) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Unable to parse replication factor for keyspace: badks"); + } + } + + @Test + public void testExtractReplicationFactorFailsLoudlyOnMissingDatacenterEntries() + { + // A NetworkTopologyStrategy keyspace with no datacenter entries is not usable. It must fail here + // rather than yielding an empty replication factor that fails later with "DC not found" + String schema = "CREATE KEYSPACE emptyks WITH REPLICATION = {'class': 'NetworkTopologyStrategy'} " + + "AND durable_writes = true;\n"; + assertThatThrownBy(() -> CqlUtils.extractReplicationFactor(schema, "emptyks")) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Unable to parse replication factor for keyspace: emptyks"); + } + @ParameterizedTest @MethodSource("org.apache.cassandra.spark.data.VersionRunner#bridges") public void testEscapedColumnNames(CassandraBridge bridge) diff --git a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java index ce51359a6..4175ad457 100644 --- a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java +++ b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java @@ -115,4 +115,20 @@ public void testSchemaBuilderWithPartiallyInitializedMetadata() new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the / form. The embedded Cassandra runs with + // transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node" for exactly the witness-enabled keyspaces the bulk + // reader needs to read. Dropping it is safe because replica placement comes from CassandraRing. + ReplicationFactor withTransient = new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + assertThat(rfToMap(withTransient)) + .containsEntry("datacenter1", "3") + .containsEntry("datacenter2", "3"); + } } diff --git a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java index 2c5a2b251..e99c00892 100644 --- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java +++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java @@ -21,6 +21,7 @@ import java.util.HashMap; +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import org.apache.cassandra.bridge.CassandraBridgeImplementation; @@ -114,4 +115,20 @@ public void testSchemaBuilderWithPartiallyInitializedMetadata() new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the / form. The embedded Cassandra runs with + // transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node" for exactly the witness-enabled keyspaces the bulk + // reader needs to read. Dropping it is safe because replica placement comes from CassandraRing. + ReplicationFactor withTransient = new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + assertThat(rfToMap(withTransient)) + .containsEntry("datacenter1", "3") + .containsEntry("datacenter2", "3"); + } } diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java index 5d0493a1f..736f89a04 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java @@ -586,6 +586,10 @@ static Map rfToMap(ReplicationFactor replicationFactor) result.put("class", "org.apache.cassandra.locator." + replicationFactor.getReplicationStrategy().name()); for (Map.Entry entry : replicationFactor.getOptions().entrySet()) { + // Deliberately emits only the total, never the / form. The embedded Cassandra runs + // with transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node". Safe to drop because replica placement comes from + // CassandraRing, not from these KeyspaceParams. result.put(entry.getKey(), Integer.toString(entry.getValue())); } return result; From 7ee268c5e6b1696dfbffaa16f2465313a8dfbea1 Mon Sep 17 00:00:00 2001 From: mkhara Date: Tue, 8 Sep 2026 13:11:28 -0700 Subject: [PATCH 03/11] Support Cassandra-reported token ranges in CassandraRing CassandraRing derives token ranges locally from tokens and the replication factor. That derivation ignores racks, while Cassandra's replica assignment is rack aware, so the derived ranges are only guaranteed to match Cassandra for a single rack per datacenter. Witness replica support needs replica information per real token range, so the derived ranges cannot be relied on. Adds an optional range-to-replicas mapping. When supplied, init() uses it instead of deriving; when absent the behaviour is unchanged, so untracked reads and the CDC caller are unaffected. Both hand-rolled serializers carry the new field and the JDK format version is bumped to 2. This is groundwork only. Nothing supplies the mapping yet. --- .../spark/data/partitioner/CassandraRing.java | 283 ++++++++++++++++-- .../data/partitioner/CassandraRingTests.java | 166 +++++++++- 2 files changed, 416 insertions(+), 33 deletions(-) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java index 752ea65f6..bf525bcf2 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java @@ -79,14 +79,24 @@ public class CassandraRing implements Serializable /** * Incremented whenever the hand-rolled JDK serialization format below changes. Version 1 added the - * per-datacenter transient (witness) replica counts after the replication options. + * per-datacenter transient (witness) replica counts after the replication options. Version 2 added the + * optional Cassandra-reported token ranges after the instances. */ - private static final byte SERIALIZATION_FORMAT_VERSION = 1; + private static final byte SERIALIZATION_FORMAT_VERSION = 2; private Partitioner partitioner; private String keyspace; private ReplicationFactor replicationFactor; private List instances; + /** + * Token ranges supplied by Cassandra rather than derived locally from tokens and replication factor. + * {@code null} means derive, which is the behaviour for every caller that predates witness replica support. + *

+ * Deriving assumes racks are not in use (see the class javadoc), whereas Cassandra's replica assignment is + * rack aware, so the derived ranges are only guaranteed to match for a single rack per datacenter. Supplying + * the ranges removes that assumption. + */ + private List explicitRanges; private transient RangeMap> replicas; private transient Multimap> tokenRangeMap; @@ -124,6 +134,26 @@ public CassandraRing(Partitioner partitioner, String keyspace, ReplicationFactor replicationFactor, Collection instances) + { + this(partitioner, keyspace, replicationFactor, instances, null); + } + + /** + * Constructs a ring from token ranges reported by Cassandra, instead of deriving them from tokens and the + * replication factor. + * + * @param partitioner the partitioner + * @param keyspace the keyspace this ring describes + * @param replicationFactor the keyspace replication factor + * @param instances all instances in the ring + * @param rangeReplicas range to replicas mapping as reported by Cassandra. Replicas must be present in + * {@code instances}. Ranges must be open-closed and non-wrapping. + */ + public CassandraRing(Partitioner partitioner, + String keyspace, + ReplicationFactor replicationFactor, + Collection instances, + Map, ? extends Collection> rangeReplicas) { this.partitioner = partitioner; this.keyspace = keyspace; @@ -131,41 +161,97 @@ public CassandraRing(Partitioner partitioner, this.instances = instances.stream() .sorted(Comparator.comparing(instance -> new BigInteger(instance.token()))) .collect(Collectors.toCollection(ArrayList::new)); + this.explicitRanges = rangeReplicas == null ? null : toExplicitRanges(rangeReplicas, this.instances); this.init(); } + /** + * Flattens the supplied mapping into a serializable form, resolving each replica to its index in + * {@code sortedInstances} so instance details are not duplicated per range. + */ + private static List toExplicitRanges( + Map, ? extends Collection> rangeReplicas, + List sortedInstances) + { + Map indexByInstance = new HashMap<>(sortedInstances.size()); + for (int index = 0; index < sortedInstances.size(); index++) + { + indexByInstance.put(sortedInstances.get(index), index); + } + + List result = new ArrayList<>(rangeReplicas.size()); + rangeReplicas.forEach((range, replicas) -> { + Preconditions.checkArgument(range.lowerEndpoint().compareTo(range.upperEndpoint()) <= 0, + "Supplied ranges must not wrap, found %s", range); + List indexes = new ArrayList<>(replicas.size()); + for (CassandraInstance replica : replicas) + { + Integer index = indexByInstance.get(replica); + Preconditions.checkArgument(index != null, + "Replica %s for range %s is not present in the ring instances", + replica, range); + indexes.add(index); + } + result.add(new ExplicitRange(range.lowerEndpoint(), range.upperEndpoint(), indexes)); + }); + result.sort(Comparator.comparing(r -> r.lower)); + return result; + } + + /** + * @return {@code true} if this ring uses token ranges reported by Cassandra rather than locally derived ones + */ + public boolean hasExplicitRanges() + { + return explicitRanges != null; + } + private void init() { // Setup token range map replicas = TreeRangeMap.create(); tokenRangeMap = ArrayListMultimap.create(); - // Calculate instance to token ranges mapping - switch (replicationFactor.getReplicationStrategy()) + if (explicitRanges != null) { - case SimpleStrategy: - tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(instances, - replicationFactor.getTotalReplicationFactor(), - partitioner)); - break; - case NetworkTopologyStrategy: - for (String dataCenter : dataCenters()) + for (ExplicitRange explicitRange : explicitRanges) + { + Range range = Range.openClosed(explicitRange.lower, explicitRange.upper); + for (int index : explicitRange.replicaIndexes) { - int rf = replicationFactor.getOptions().get(dataCenter); - if (rf == 0) + tokenRangeMap.put(instances.get(index), range); + } + } + } + else + { + // Calculate instance to token ranges mapping + switch (replicationFactor.getReplicationStrategy()) + { + case SimpleStrategy: + tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(instances, + replicationFactor.getTotalReplicationFactor(), + partitioner)); + break; + case NetworkTopologyStrategy: + for (String dataCenter : dataCenters()) { - continue; + int rf = replicationFactor.getOptions().get(dataCenter); + if (rf == 0) + { + continue; + } + List dcInstances = instances.stream() + .filter(instance -> instance.dataCenter().matches(dataCenter)) + .collect(Collectors.toList()); + tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(dcInstances, + replicationFactor.getOptions().get(dataCenter), + partitioner)); } - List dcInstances = instances.stream() - .filter(instance -> instance.dataCenter().matches(dataCenter)) - .collect(Collectors.toList()); - tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(dcInstances, - replicationFactor.getOptions().get(dataCenter), - partitioner)); - } - break; - default: - throw new UnsupportedOperationException("Unsupported replication strategy"); + break; + default: + throw new UnsupportedOperationException("Unsupported replication strategy"); + } } // Calculate token range to replica mapping @@ -173,6 +259,55 @@ private void init() tokenRangeMap.asMap().forEach((instance, ranges) -> ranges.forEach(range -> addReplica(instance, range, replicas))); } + /** + * A single Cassandra-reported token range and the instances replicating it, held by index into + * {@link CassandraRing#instances}. + */ + private static final class ExplicitRange implements Serializable + { + private static final long serialVersionUID = 2026090800000000001L; + + private final BigInteger lower; + private final BigInteger upper; + private final List replicaIndexes; + + private ExplicitRange(BigInteger lower, BigInteger upper, List replicaIndexes) + { + this.lower = lower; + this.upper = upper; + this.replicaIndexes = replicaIndexes; + } + + @Override + public boolean equals(Object other) + { + if (this == other) + { + return true; + } + if (other == null || getClass() != other.getClass()) + { + return false; + } + ExplicitRange that = (ExplicitRange) other; + return Objects.equals(lower, that.lower) + && Objects.equals(upper, that.upper) + && Objects.equals(replicaIndexes, that.replicaIndexes); + } + + @Override + public int hashCode() + { + return Objects.hash(lower, upper, replicaIndexes); + } + + @Override + public String toString() + { + return "(" + lower + ", " + upper + "]=" + replicaIndexes; + } + } + public Partitioner partitioner() { return partitioner; @@ -261,6 +396,7 @@ public boolean equals(Object other) && Objects.equals(this.keyspace, that.keyspace) && Objects.equals(this.replicationFactor, that.replicationFactor) && Objects.equals(this.instances, that.instances) + && Objects.equals(this.explicitRanges, that.explicitRanges) && Objects.equals(this.replicas, that.replicas) && Objects.equals(this.tokenRangeMap, that.tokenRangeMap); } @@ -268,7 +404,7 @@ public boolean equals(Object other) @Override public int hashCode() { - return Objects.hash(partitioner, keyspace, replicationFactor, instances, replicas, tokenRangeMap); + return Objects.hash(partitioner, keyspace, replicationFactor, instances, explicitRanges, replicas, tokenRangeMap); } private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException @@ -304,6 +440,29 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE { this.instances.add(new CassandraInstance(in.readUTF(), in.readUTF(), in.readUTF())); } + + int numExplicitRanges = in.readInt(); + if (numExplicitRanges < 0) + { + this.explicitRanges = null; + } + else + { + List ranges = new ArrayList<>(numExplicitRanges); + for (int range = 0; range < numExplicitRanges; range++) + { + BigInteger lower = new BigInteger(in.readUTF()); + BigInteger upper = new BigInteger(in.readUTF()); + int numReplicas = in.readShort(); + List indexes = new ArrayList<>(numReplicas); + for (int replica = 0; replica < numReplicas; replica++) + { + indexes.add((int) in.readShort()); + } + ranges.add(new ExplicitRange(lower, upper, indexes)); + } + this.explicitRanges = ranges; + } this.init(); } @@ -337,6 +496,26 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou out.writeUTF(instance.nodeName()); out.writeUTF(instance.dataCenter()); } + + // -1 distinguishes "derive the ranges" from "explicitly zero ranges" + if (this.explicitRanges == null) + { + out.writeInt(-1); + } + else + { + out.writeInt(this.explicitRanges.size()); + for (ExplicitRange range : this.explicitRanges) + { + out.writeUTF(range.lower.toString()); + out.writeUTF(range.upper.toString()); + out.writeShort(range.replicaIndexes.size()); + for (int index : range.replicaIndexes) + { + out.writeShort(index); + } + } + } } public static class Serializer extends com.esotericsoftware.kryo.Serializer @@ -348,17 +527,61 @@ public void write(Kryo kryo, Output out, CassandraRing ring) out.writeString(ring.keyspace); kryo.writeObject(out, ring.replicationFactor); kryo.writeObject(out, ring.instances); + // -1 distinguishes "derive the ranges" from "explicitly zero ranges" + if (ring.explicitRanges == null) + { + out.writeInt(-1); + } + else + { + out.writeInt(ring.explicitRanges.size()); + for (ExplicitRange range : ring.explicitRanges) + { + out.writeString(range.lower.toString()); + out.writeString(range.upper.toString()); + out.writeInt(range.replicaIndexes.size()); + for (int index : range.replicaIndexes) + { + out.writeInt(index); + } + } + } } @Override @SuppressWarnings("unchecked") public CassandraRing read(Kryo kryo, Input in, Class type) { - return new CassandraRing(in.readByte() == 1 ? Partitioner.RandomPartitioner - : Partitioner.Murmur3Partitioner, - in.readString(), - kryo.readObject(in, ReplicationFactor.class), - kryo.readObject(in, ArrayList.class)); + Partitioner partitioner = in.readByte() == 1 ? Partitioner.RandomPartitioner + : Partitioner.Murmur3Partitioner; + String keyspace = in.readString(); + ReplicationFactor replicationFactor = kryo.readObject(in, ReplicationFactor.class); + List instances = kryo.readObject(in, ArrayList.class); + + int numExplicitRanges = in.readInt(); + if (numExplicitRanges < 0) + { + return new CassandraRing(partitioner, keyspace, replicationFactor, instances); + } + + // Instances are sorted by the constructor, and were written in that order, so indexes still resolve + List sorted = instances.stream() + .sorted(Comparator.comparing(i -> new BigInteger(i.token()))) + .collect(Collectors.toList()); + Map, List> rangeReplicas = new HashMap<>(numExplicitRanges); + for (int range = 0; range < numExplicitRanges; range++) + { + BigInteger lower = new BigInteger(in.readString()); + BigInteger upper = new BigInteger(in.readString()); + int numReplicas = in.readInt(); + List replicas = new ArrayList<>(numReplicas); + for (int replica = 0; replica < numReplicas; replica++) + { + replicas.add(sorted.get(in.readInt())); + } + rangeReplicas.put(Range.openClosed(lower, upper), replicas); + } + return new CassandraRing(partitioner, keyspace, replicationFactor, instances, rangeReplicas); } } } diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java index 89cb14be3..f31a118ad 100644 --- a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java +++ b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java @@ -27,8 +27,15 @@ import java.math.BigInteger; import java.util.Arrays; import java.util.Collection; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Multimap; import com.google.common.collect.Range; @@ -522,10 +529,12 @@ public void testJdkDeserializationRejectsUnknownFormatVersion() throws Exception int keyspaceIndex = indexOf(raw, keyspaceUtf); assertThat(keyspaceIndex).as("keyspace UTF header should be locatable").isGreaterThan(1); int versionIndex = keyspaceIndex - 2; - assertThat(raw[versionIndex]).as("format version byte").isEqualTo((byte) 1); + // Deliberately not asserting the current version number, so this test does not need updating + // every time the format changes + assertThat(raw[versionIndex]).as("format version byte should be positive").isGreaterThan((byte) 0); - // Simulate a stream written before the transient-replica section existed - raw[versionIndex] = 0; + // Simulate a stream written by a version whose format we do not understand + raw[versionIndex] = (byte) 99; try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(raw))) { @@ -551,4 +560,155 @@ private static int indexOf(byte[] haystack, byte[] needle) } return -1; } + + // Cassandra-reported (explicit) token ranges, used for witness-enabled keyspaces where the locally + // derived ranges cannot be trusted because the derivation ignores racks + + private static CassandraRing explicitRangeRing() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1"), + new CassandraInstance("200", "n3", "dc1")); + // Deliberately not what the derivation would produce: boundary at 50, and only two of three + // replicas per range. A derived ring can never produce this shape. + Map, List> explicit = new LinkedHashMap<>(); + explicit.put(Range.openClosed(Partitioner.Murmur3Partitioner.minToken(), BigInteger.valueOf(50L)), + Arrays.asList(instances.get(0), instances.get(1))); + explicit.put(Range.openClosed(BigInteger.valueOf(50L), Partitioner.Murmur3Partitioner.maxToken()), + Arrays.asList(instances.get(1), instances.get(2))); + return new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "3/1")), + instances, explicit); + } + + private static Set replicaNamesAt(CassandraRing ring, long token) + { + Set names = new TreeSet<>(); + ring.getReplicas(BigInteger.valueOf(token)).forEach(instance -> names.add(instance.nodeName())); + return names; + } + + @Test + public void testExplicitRangesAreHonouredInsteadOfDerived() + { + CassandraRing explicitRing = explicitRangeRing(); + assertThat(explicitRing.hasExplicitRanges()).isTrue(); + // the supplied boundary at 50 is used, and only the supplied replicas are returned + assertThat(replicaNamesAt(explicitRing, 10L)).containsExactly("n1", "n2"); + assertThat(replicaNamesAt(explicitRing, 1000L)).containsExactly("n2", "n3"); + } + + @Test + public void testDerivedRingIsUnchangedByExplicitRangeSupport() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1"), + new CassandraInstance("200", "n3", "dc1")); + CassandraRing derived = new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "3")), + instances); + assertThat(derived.hasExplicitRanges()).isFalse(); + // derivation puts boundaries at node tokens, so 50 is not a boundary + Set upperBounds = new TreeSet<>(); + derived.rangeMap().asMapOfRanges().forEach((range, replicas) -> { + if (!replicas.isEmpty()) + { + upperBounds.add(range.upperEndpoint()); + } + }); + assertThat(upperBounds).doesNotContain(BigInteger.valueOf(50L)); + } + + @Test + public void testExplicitRangesSurviveJdkSerialization() throws Exception + { + CassandraRing original = explicitRangeRing(); + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(original); + } + CassandraRing deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (CassandraRing) in.readObject(); + } + assertThat(deserialized.hasExplicitRanges()).isTrue(); + assertThat(replicaNamesAt(deserialized, 10L)).containsExactly("n1", "n2"); + assertThat(replicaNamesAt(deserialized, 1000L)).containsExactly("n2", "n3"); + } + + @Test + public void testExplicitRangesSurviveKryoSerialization() + { + CassandraRing original = explicitRangeRing(); + Kryo kryo = new Kryo(); + kryo.register(CassandraRing.class, new CassandraRing.Serializer()); + kryo.register(ReplicationFactor.class, new ReplicationFactor.Serializer()); + kryo.register(CassandraInstance.class, new CassandraInstance.Serializer()); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (Output out = new Output(bytes)) + { + kryo.writeObject(out, original); + } + CassandraRing deserialized; + try (Input in = new Input(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = kryo.readObject(in, CassandraRing.class); + } + assertThat(deserialized.hasExplicitRanges()).isTrue(); + assertThat(replicaNamesAt(deserialized, 10L)).containsExactly("n1", "n2"); + } + + @Test + public void testDerivedRingStaysDerivedThroughKryo() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1"), + new CassandraInstance("200", "n3", "dc1")); + CassandraRing derived = new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "3")), + instances); + Kryo kryo = new Kryo(); + kryo.register(CassandraRing.class, new CassandraRing.Serializer()); + kryo.register(ReplicationFactor.class, new ReplicationFactor.Serializer()); + kryo.register(CassandraInstance.class, new CassandraInstance.Serializer()); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (Output out = new Output(bytes)) + { + kryo.writeObject(out, derived); + } + CassandraRing deserialized; + try (Input in = new Input(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = kryo.readObject(in, CassandraRing.class); + } + assertThat(deserialized.hasExplicitRanges()).isFalse(); + assertThat(deserialized).isEqualTo(derived); + } + + @Test + public void testExplicitRangeWithUnknownReplicaIsRejected() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1")); + Map, List> explicit = new LinkedHashMap<>(); + explicit.put(Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(100L)), + Arrays.asList(new CassandraInstance("999", "nope", "dc1"))); + assertThatThrownBy(() -> new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "2")), + instances, explicit)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("not present in the ring instances"); + } } From a91e0a15203a2101f88b5cf9bcc04b63dcc6e0e2 Mon Sep 17 00:00:00 2001 From: mkhara Date: Tue, 8 Sep 2026 13:22:01 -0700 Subject: [PATCH 04/11] Source token ranges from Cassandra for mutation tracked keyspaces For a tracked keyspace a range may be replicated to a witness that holds no data, so the reader must know which instance replicates which range. The locally derived ranges cannot answer that: the derivation ignores racks while Cassandra's replica assignment is rack aware, so the two only agree for a single rack per datacenter. When the keyspace is tracked, the range to replica mapping is now taken from the token-range-replicas endpoint (readReplicas, so pending ranges are excluded). Untracked keyspaces keep using the ring response exactly as before. Node discovery still comes from the ring response in both cases, so snapshot creation, sizing and the Sidecar client pool are untouched. Only the range to replica mapping changes. Replicas are reported as "address:port" and instances are keyed by fqdn, so the join goes through the replica metadata. A replica absent from the ring is skipped with a warning, since the ring is the source of truth for which nodes the reader can reach. A node owning multiple tokens is rejected, because it cannot be represented as a single CassandraInstance; mutation tracking requires num_tokens=1 so this should not arise. Adds a forceCassandraTokenRanges option to exercise the new path against untracked keyspaces. Tracked keyspaces need Cassandra 6.0 bridge modules before they can be created in integration tests, so without this the path would merge with no integration coverage. --- .../spark/data/CassandraDataLayer.java | 109 +++++++++++++++++- .../cassandra/spark/data/ClientConfig.java | 17 +++ 2 files changed, 125 insertions(+), 1 deletion(-) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java index 99cbc6827..4cb50d252 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java @@ -62,6 +62,7 @@ import o.a.c.sidecar.client.shaded.common.response.NodeSettings; import o.a.c.sidecar.client.shaded.common.response.RingResponse; import o.a.c.sidecar.client.shaded.common.response.SchemaResponse; +import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse; import org.apache.cassandra.analytics.stats.Stats; import org.apache.cassandra.bridge.BigNumberConfig; import org.apache.cassandra.bridge.BigNumberConfigImpl; @@ -317,7 +318,25 @@ private int initBulkReader(@NotNull ClientConfig options) throws ExecutionExcept udts.forEach(udt -> LOGGER.info("Adding schema UDT: '{}'", udt)); cqlTable = bridge().buildSchema(createStmt, keyspace, replicationFactor, partitioner, udts, null, indexCount, false); - CassandraRing ring = createCassandraRingFromRing(partitioner, replicationFactor, ringFuture.get()); + + // Mutation tracked keyspaces may replicate a range to a witness that holds no data, so the range to replica + // mapping must come from Cassandra rather than being derived locally. See + // createCassandraRingFromTokenRangeReplicas. + String replicationType = CqlUtils.extractReplicationType(fullSchema, keyspace); + boolean tracked = bridge().isTracked(replicationType); + CassandraRing ring; + if (tracked || options.forceCassandraTokenRanges()) + { + LOGGER.info("Sourcing token ranges from Cassandra keyspace={} replicationType={} forced={}", + keyspace, replicationType, options.forceCassandraTokenRanges()); + TokenRangeReplicasResponse topology = sidecar.tokenRangeReplicas(new ArrayList<>(clusterConfig), + maybeQuotedKeyspace).get(); + ring = createCassandraRingFromTokenRangeReplicas(partitioner, replicationFactor, ringFuture.get(), topology); + } + else + { + ring = createCassandraRingFromRing(partitioner, replicationFactor, ringFuture.get()); + } int effectiveNumberOfCores = sizingFuture.get(); tokenPartitioner = new TokenPartitioner(ring, options.defaultParallelism(), effectiveNumberOfCores); @@ -691,6 +710,94 @@ public CassandraRing createCassandraRingFromRing(Partitioner partitioner, return new CassandraRing(partitioner, keyspace, replicationFactor, instances); } + /** + * Builds the ring using the token ranges Cassandra reports, rather than deriving them from tokens and the + * replication factor. The local derivation assumes racks are not in use, whereas Cassandra's replica assignment + * is rack aware, so for mutation tracked keyspaces - where a replica may be a witness holding no data - the + * derived ranges cannot be relied on to identify which instance replicates which range. + *

+ * Node discovery still comes from the ring response, so snapshot creation, sizing and the Sidecar client pool + * are unaffected. Only the range to replica mapping is taken from Cassandra. + * + * @param partitioner the partitioner + * @param replicationFactor the keyspace replication factor + * @param ring the ring response, used for node discovery + * @param topology the token range replicas response, used for the range to replica mapping + * @return a ring whose token ranges were reported by Cassandra + */ + public CassandraRing createCassandraRingFromTokenRangeReplicas(Partitioner partitioner, + ReplicationFactor replicationFactor, + RingResponse ring, + TokenRangeReplicasResponse topology) + { + List instances = ring + .stream() + .filter(status -> datacenter == null || datacenter.equalsIgnoreCase(status.datacenter())) + .map(status -> new CassandraInstance(status.token(), status.fqdn(), status.datacenter())) + .collect(Collectors.toList()); + + // Token range replicas identify replicas by "address:port", while instances built from the ring are keyed + // by fqdn, so the replica metadata is used to translate between them + Map instanceByNodeName = new HashMap<>(instances.size()); + for (CassandraInstance instance : instances) + { + CassandraInstance previous = instanceByNodeName.put(instance.nodeName(), instance); + if (previous != null && !previous.equals(instance)) + { + // A node owning several tokens cannot be represented as a single CassandraInstance. Mutation + // tracking requires num_tokens=1, so this should not happen on a tracked keyspace's cluster + throw new IllegalStateException(String.format( + "Node %s owns multiple tokens (%s and %s). Sourcing token ranges from Cassandra requires " + + "single-token nodes.", instance.nodeName(), previous.token(), instance.token())); + } + } + + Map, List> rangeReplicas = new HashMap<>(); + for (TokenRangeReplicasResponse.ReplicaInfo replicaInfo : topology.readReplicas()) + { + Range range = Range.openClosed(new BigInteger(replicaInfo.start()), + new BigInteger(replicaInfo.end())); + List replicas = new ArrayList<>(); + for (Map.Entry> byDatacenter : replicaInfo.replicasByDatacenter().entrySet()) + { + if (datacenter != null && !datacenter.equalsIgnoreCase(byDatacenter.getKey())) + { + continue; + } + for (String replicaKey : byDatacenter.getValue()) + { + TokenRangeReplicasResponse.ReplicaMetadata metadata = topology.replicaMetadata().get(replicaKey); + String nodeName = metadata != null ? metadata.fqdn() : replicaKey; + CassandraInstance instance = instanceByNodeName.get(nodeName); + if (instance == null) + { + // The ring response is the source of truth for which nodes the reader can talk to, so a + // replica absent from it is skipped rather than failing the job + LOGGER.warn("Skipping replica reported by Cassandra that is absent from the ring " + + "replica={} fqdn={} range=({}, {}]", + replicaKey, nodeName, replicaInfo.start(), replicaInfo.end()); + continue; + } + replicas.add(instance); + } + } + if (!replicas.isEmpty()) + { + rangeReplicas.put(range, replicas); + } + } + + if (rangeReplicas.isEmpty()) + { + throw new IllegalStateException( + "Cassandra reported no read replicas for keyspace " + keyspace + " in datacenter " + datacenter); + } + + LOGGER.info("Using token ranges reported by Cassandra keyspace={} numRanges={} numInstances={}", + keyspace, rangeReplicas.size(), instances.size()); + return new CassandraRing(partitioner, keyspace, replicationFactor, instances, rangeReplicas); + } + // Startup Validation @Override diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java index 4ca6c9214..8d35fbd2d 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java @@ -55,6 +55,13 @@ public class ClientConfig public static final String SNAPSHOT_NAME_KEY = "snapshotName"; public static final String DC_KEY = "dc"; public static final String CREATE_SNAPSHOT_KEY = "createSnapshot"; + /** + * Forces token ranges to be sourced from Cassandra via the token-range-replicas endpoint even for keyspaces + * that are not mutation tracked. Normally this happens only for tracked keyspaces. Exists so the code path can + * be exercised against ordinary clusters, which is the only coverage available until Cassandra 6.0 bridge + * modules land and tracked keyspaces can be created in integration tests. + */ + public static final String FORCE_CASSANDRA_TOKEN_RANGES_KEY = "forceCassandraTokenRanges"; public static final String CLEAR_SNAPSHOT_KEY = "clearSnapshot"; /** * Format of clearSnapshotStrategy is {strategy [snapshotTTLvalue]}, clearSnapshotStrategy holds both the strategy @@ -101,6 +108,7 @@ public class ClientConfig protected String snapshotName; protected String datacenter; protected boolean createSnapshot; + protected boolean forceCassandraTokenRanges; protected boolean clearSnapshot; protected ClearSnapshotStrategy clearSnapshotStrategy; protected int defaultParallelism; @@ -127,6 +135,7 @@ protected ClientConfig(Map options) this.snapshotName = MapUtils.getOrDefault(options, SNAPSHOT_NAME_KEY, "sbr_" + UUID.randomUUID().toString().replace("-", "")); this.datacenter = options.get(MapUtils.lowerCaseKey(DC_KEY)); this.createSnapshot = MapUtils.getBoolean(options, CREATE_SNAPSHOT_KEY, true); + this.forceCassandraTokenRanges = MapUtils.getBoolean(options, FORCE_CASSANDRA_TOKEN_RANGES_KEY, false); this.clearSnapshot = MapUtils.getBoolean(options, CLEAR_SNAPSHOT_KEY, createSnapshot); String clearSnapshotStrategyOption = MapUtils.getOrDefault(options, CLEAR_SNAPSHOT_STRATEGY_KEY, null); @@ -204,6 +213,14 @@ public String datacenter() return datacenter; } + /** + * @return {@code true} to source token ranges from Cassandra regardless of whether the keyspace is tracked + */ + public boolean forceCassandraTokenRanges() + { + return forceCassandraTokenRanges; + } + public boolean createSnapshot() { return createSnapshot; From ea80f77c73d811177ad0b02673dfe51de6809b4f Mon Sep 17 00:00:00 2001 From: mkhara Date: Tue, 8 Sep 2026 13:28:48 -0700 Subject: [PATCH 05/11] Test sourcing token ranges from Cassandra Covers createCassandraRingFromTokenRangeReplicas: the address-to-fqdn join through the replica metadata, datacenter filtering, skipping a replica absent from the ring, rejecting a node that owns multiple tokens, and failing when Cassandra reports no read replicas. The fixture uses a range boundary the local derivation cannot produce, so the tests fail if the Cassandra-reported ranges are ignored. Also documents that forceCassandraTokenRanges is set lowercased, since Spark lowercases option keys and MapUtils looks them up that way. --- .../cassandra/spark/data/ClientConfig.java | 2 + .../spark/data/CassandraDataLayerTests.java | 185 ++++++++++++++++++ 2 files changed, 187 insertions(+) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java index 8d35fbd2d..89463da7f 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java @@ -60,6 +60,8 @@ public class ClientConfig * that are not mutation tracked. Normally this happens only for tracked keyspaces. Exists so the code path can * be exercised against ordinary clusters, which is the only coverage available until Cassandra 6.0 bridge * modules land and tracked keyspaces can be created in integration tests. + *

+ * Set as {@code forcecassandratokenranges}; Spark lowercases option keys. */ public static final String FORCE_CASSANDRA_TOKEN_RANGES_KEY = "forceCassandraTokenRanges"; public static final String CLEAR_SNAPSHOT_KEY = "clearSnapshot"; diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java index 088de9070..b69c6eb5b 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java @@ -19,15 +19,31 @@ package org.apache.cassandra.spark.data; +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 java.util.stream.Collectors; import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import o.a.c.sidecar.client.shaded.common.response.RingResponse; +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 o.a.c.sidecar.client.shaded.common.response.data.RingEntry; +import org.apache.cassandra.clients.Sidecar; +import org.apache.cassandra.spark.data.partitioner.CassandraInstance; +import org.apache.cassandra.spark.data.partitioner.CassandraRing; +import org.apache.cassandra.spark.data.partitioner.Partitioner; + import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class CassandraDataLayerTests { @@ -64,4 +80,173 @@ void testClearSnapshotOptionSupport(Boolean clearSnapshot, String expectedClearS assertThat(clearSnapshotStrategy.hasTTL()).isEqualTo(expectedClearSnapshotStrategy.hasTTL()); assertThat(clearSnapshotStrategy.ttl()).isEqualTo(expectedClearSnapshotStrategy.ttl()); } + + // Sourcing token ranges from Cassandra, used for mutation tracked keyspaces where the locally derived + // ranges cannot identify which instance replicates which range + + private static final String KS = "big-data"; + + private static CassandraDataLayer dataLayer(String datacenter) + { + Map options = new HashMap<>(REQUIRED_CLIENT_CONFIG_OPTIONS); + if (datacenter != null) + { + options.put("dc", datacenter); + } + return new CassandraDataLayer(ClientConfig.create(options), Sidecar.ClientConfig.create(), null); + } + + private static RingResponse ringOf(String... fqdnAndTokenAndDc) + { + RingResponse ring = new RingResponse(); + for (int i = 0; i < fqdnAndTokenAndDc.length; i += 3) + { + ring.add(new RingEntry.Builder().fqdn(fqdnAndTokenAndDc[i]) + .token(fqdnAndTokenAndDc[i + 1]) + .datacenter(fqdnAndTokenAndDc[i + 2]) + .address(fqdnAndTokenAndDc[i]) + .port(9042) + .rack("rack1") + .status("UP") + .state("NORMAL") + .load("1") + .owns("1") + .hostId("h" + i) + .build()); + } + return ring; + } + + /** + * Replica keys are "address:port" and are translated to fqdn through the replica metadata, mirroring what + * Sidecar returns. + */ + private static TokenRangeReplicasResponse topologyOf(List readReplicas, String... addressToFqdn) + { + Map metadata = new HashMap<>(); + for (int i = 0; i < addressToFqdn.length; i += 2) + { + String key = addressToFqdn[i]; + String fqdn = addressToFqdn[i + 1]; + metadata.put(key, new ReplicaMetadata("NORMAL", "UP", fqdn, key.split(":")[0], 9042, "dc1")); + } + return new TokenRangeReplicasResponse(readReplicas, readReplicas, metadata); + } + + @Test + void testRingFromTokenRangeReplicasUsesCassandraReportedRanges() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1", "n3", "200", "dc1"); + + // boundary at 50 with only two replicas: a shape the local derivation cannot produce + List readReplicas = Arrays.asList( + new ReplicaInfo("-9223372036854775808", "50", ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042"))), + new ReplicaInfo("50", "9223372036854775807", ImmutableMap.of("dc1", Arrays.asList("1.1.1.2:9042", "1.1.1.3:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2", + "1.1.1.3:9042", "n3"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(3), ring, topology); + + assertThat(result.hasExplicitRanges()).isTrue(); + assertThat(replicaNames(result, 10L)).containsExactly("n1", "n2"); + assertThat(replicaNames(result, 1000L)).containsExactly("n2", "n3"); + } + + @Test + void testRingFromTokenRangeReplicasFiltersByDatacenter() + { + CassandraDataLayer layer = dataLayer("dc1"); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1", "n4", "300", "dc2"); + + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042"), + "dc2", Collections.singletonList("1.1.1.4:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2", + "1.1.1.4:9042", "n4"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology); + + // the dc2 replica must not appear, even though Cassandra reported it + assertThat(replicaNames(result, 10L)).containsExactly("n1", "n2"); + } + + @Test + void testRingFromTokenRangeReplicasSkipsReplicaAbsentFromRing() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.9:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.9:9042", "ghost"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology); + + // the ring response is the source of truth for reachable nodes, so the unknown replica is dropped + assertThat(replicaNames(result, 10L)).containsExactly("n1"); + } + + @Test + void testRingFromTokenRangeReplicasRejectsMultiTokenNode() + { + CassandraDataLayer layer = dataLayer(null); + // same fqdn twice with different tokens, i.e. vnodes + RingResponse ring = ringOf("n1", "0", "dc1", "n1", "100", "dc1"); + + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Collections.singletonList("1.1.1.1:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, "1.1.1.1:9042", "n1"); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(1), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("owns multiple tokens"); + } + + @Test + void testRingFromTokenRangeReplicasFailsWhenNoReadReplicas() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1"); + TokenRangeReplicasResponse topology = topologyOf(Collections.emptyList()); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(1), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("no read replicas"); + } + + @Test + void testForceCassandraTokenRangesDefaultsToFalse() + { + assertThat(ClientConfig.create(new HashMap<>(REQUIRED_CLIENT_CONFIG_OPTIONS)) + .forceCassandraTokenRanges()).isFalse(); + + // Spark lowercases option keys, and MapUtils looks them up lowercased, so this is how a user sets it + Map options = new HashMap<>(REQUIRED_CLIENT_CONFIG_OPTIONS); + options.put("forcecassandratokenranges", "true"); + assertThat(ClientConfig.create(options).forceCassandraTokenRanges()).isTrue(); + } + + private static List replicaNames(CassandraRing ring, long token) + { + return ring.getReplicas(BigInteger.valueOf(token)) + .stream() + .map(CassandraInstance::nodeName) + .sorted() + .collect(Collectors.toList()); + } } From 57dfa7b0ddf7b700ce097c8801115174dbdcee55 Mon Sep 17 00:00:00 2001 From: mkhara Date: Tue, 8 Sep 2026 14:51:30 -0700 Subject: [PATCH 06/11] Integration tests for Cassandra-supplied token ranges, including a rack-aware cluster BulkReaderCassandraTokenRangesTest exercises the new topology path against an ordinary keyspace using forcecassandratokenranges, since tracked keyspaces cannot be created until Cassandra 6.0 bridge modules land. It asserts the full dataset rather than a row count, because an incomplete read is the failure mode the path exists to prevent, and covers LOCAL_QUORUM and a pushdown filter. BulkReaderMultiRackTest closes a coverage gap. Every existing test places all nodes of a datacenter in one rack, where Cassandra's rack-aware assignment reduces to taking the next RF nodes in ring order, which is what the reader derives. The cluster here uses six nodes across three racks with two adjacent nodes per rack and RF 3, so ring order starts 1,2,3 while Cassandra must pick 1,3,5 to satisfy one replica per rack. The two topology sources therefore genuinely disagree. One node per rack, or round-robin racks, would coincide and prove nothing. CassandraCluster only applied dcAndRackSupplier when dcCount was greater than one, so a single-datacenter multi-rack topology was silently ignored and fell back to the default rack. An explicit supplier is now always honoured; the built-in two-datacenter default is unchanged. --- .../distributed/impl/CassandraCluster.java | 23 ++- .../BulkReaderCassandraTokenRangesTest.java | 154 ++++++++++++++++ .../analytics/BulkReaderMultiRackTest.java | 167 ++++++++++++++++++ 3 files changed, 332 insertions(+), 12 deletions(-) create mode 100644 cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderCassandraTokenRangesTest.java create mode 100644 cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java 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 93a877ea3..585c49a7d 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 @@ -137,19 +137,18 @@ public AbstractCluster initializeCluster(String versionString, clusterBuilder.withTokenSupplier(tokenSupplier) .withConfig(instanceConfigUpdater); - if (dcCount > 1) + // An explicit supplier is always honoured, including for a single datacenter, so that a test can place + // nodes of one datacenter in different racks. The built-in default only makes sense for multiple datacenters. + if (configuration.dcAndRackSupplier != null) { - if (configuration.dcAndRackSupplier != null) - { - clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, configuration.dcAndRackSupplier)); - } - else - { - clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, - (nodeId) -> nodeId % 2 != 0 ? - dcAndRack("datacenter1", "rack1") : - dcAndRack("datacenter2", "rack2"))); - } + clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, configuration.dcAndRackSupplier)); + } + else if (dcCount > 1) + { + clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, + (nodeId) -> nodeId % 2 != 0 ? + dcAndRack("datacenter1", "rack1") : + dcAndRack("datacenter2", "rack2"))); } if (configuration.instanceInitializer != null) diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderCassandraTokenRangesTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderCassandraTokenRangesTest.java new file mode 100644 index 000000000..d81529a98 --- /dev/null +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderCassandraTokenRangesTest.java @@ -0,0 +1,154 @@ +/* + * 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 java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.sidecar.testing.QualifiedName; +import org.apache.cassandra.testing.ClusterBuilderConfiguration; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; + +import static org.apache.cassandra.testing.TestUtils.DC1_RF3; +import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE; +import static org.apache.cassandra.testing.TestUtils.uniqueTestTableFullName; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests bulk reads that source token ranges from Cassandra via the token-range-replicas endpoint, rather than + * deriving them locally from tokens and the replication factor. + *

+ * Mutation tracked keyspaces take that path automatically, because a range may be replicated to a witness holding + * no data and the local derivation cannot say which instance replicates which range. Tracked keyspaces cannot be + * created until Cassandra 6.0 bridge modules land (CASSANALYTICS-192), so these tests use the + * {@code forcecassandratokenranges} option to exercise the same path against an ordinary keyspace. + *

+ * Reads must return exactly the same data either way, so each test asserts the full dataset rather than only a row + * count - an incomplete read is the failure mode this path exists to prevent. + */ +class BulkReaderCassandraTokenRangesTest extends SharedClusterSparkIntegrationTestBase +{ + static final List DATASET = Arrays.asList("a", "b", "c", "d", "e", "f", "g"); + + QualifiedName table = uniqueTestTableFullName(TEST_KEYSPACE); + + @Override + protected ClusterBuilderConfiguration testClusterConfiguration() + { + return super.testClusterConfiguration() + .nodesPerDc(3); + } + + @Test + void testReadWithCassandraSuppliedTokenRanges() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load(); + assertCompleteDataset(data); + } + + /** + * The derived and Cassandra-supplied paths must agree. This is the check that would catch the ranges being + * misaligned, which for a witness-enabled keyspace would show up as silently missing rows. + */ + @Test + void testCassandraSuppliedRangesMatchDerivedRanges() + { + Dataset derived = bulkReaderDataFrame(table).load(); + Dataset supplied = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load(); + + assertCompleteDataset(derived); + assertCompleteDataset(supplied); + assertThat(sortedNames(supplied)) + .describedAs("reads with Cassandra-supplied ranges must return the same rows as derived ranges") + .isEqualTo(sortedNames(derived)); + } + + @Test + void testReadWithCassandraSuppliedTokenRangesAtQuorum() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .option("consistencyLevel", "LOCAL_QUORUM") + .load(); + assertCompleteDataset(data); + } + + /** + * Pushdown filters are applied against the Spark partition token ranges, which are derived from the ring's + * ranges, so they are worth exercising on the new path. + */ + @Test + void testPushDownFilterWithCassandraSuppliedTokenRanges() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load() + .filter("id = 3"); + + List rows = data.collectAsList(); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getInt(0)).isEqualTo(3); + assertThat(rows.get(0).getString(1)).isEqualTo(DATASET.get(3)); + } + + private void assertCompleteDataset(Dataset data) + { + List rows = data.collectAsList().stream() + .sorted(Comparator.comparing(row -> row.getInt(0))) + .collect(Collectors.toList()); + assertThat(rows).hasSize(DATASET.size()); + for (int i = 0; i < DATASET.size(); i++) + { + assertThat(rows.get(i).getInt(0)).isEqualTo(i); + assertThat(rows.get(i).getString(1)).isEqualTo(DATASET.get(i)); + } + } + + private List sortedNames(Dataset data) + { + return data.collectAsList().stream() + .map(row -> row.getInt(0) + "=" + row.getString(1)) + .sorted() + .collect(Collectors.toList()); + } + + @Override + protected void initializeSchemaForTest() + { + createTestKeyspace(TEST_KEYSPACE, DC1_RF3); + createTestTable(table, "CREATE TABLE IF NOT EXISTS %s (id int PRIMARY KEY, name text);"); + + IInstance firstRunningInstance = cluster.getFirstRunningInstance(); + for (int i = 0; i < DATASET.size(); i++) + { + firstRunningInstance.coordinator() + .execute(String.format("INSERT INTO %s (id, name) VALUES (%d, '%s');", + table, i, DATASET.get(i)), + ConsistencyLevel.ALL); + } + } +} diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java new file mode 100644 index 000000000..9576dee8e --- /dev/null +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java @@ -0,0 +1,167 @@ +/* + * 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 java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.sidecar.testing.QualifiedName; +import org.apache.cassandra.testing.ClusterBuilderConfiguration; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; + +import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack; +import static org.apache.cassandra.testing.TestUtils.DC1_RF3; +import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE; +import static org.apache.cassandra.testing.TestUtils.uniqueTestTableFullName; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests bulk reads against a datacenter split across several racks. + *

+ * This closes a coverage gap. Every other integration test places all nodes of a datacenter in a single rack, where + * Cassandra's rack-aware replica assignment reduces to "take the next RF nodes in ring order" - exactly what the + * reader derives locally. With one rack per replica, {@code NetworkTopologyStrategy} instead skips nodes to spread + * replicas across racks (see {@code acceptableRackRepeats}), so a derived replica set can disagree with Cassandra's. + *

+ * {@code CassandraRing} states that it assumes racks are not in use, so a divergence here would previously have been + * invisible in CI and would surface in production as a bulk read silently returning incomplete data. + */ +class BulkReaderMultiRackTest extends SharedClusterSparkIntegrationTestBase +{ + static final List DATASET = Arrays.asList("a", "b", "c", "d", "e", "f", "g"); + + QualifiedName table = uniqueTestTableFullName(TEST_KEYSPACE); + + @Override + protected ClusterBuilderConfiguration testClusterConfiguration() + { + // Six nodes, three racks, two adjacent nodes per rack, with RF 3. + // + // The pairing matters. Tokens are handed out in node order, so "the next RF nodes in ring order" - what the + // reader derives - starts 1,2,3. Cassandra cannot use that, because nodes 1 and 2 share rack1, so it skips + // to satisfy one replica per rack and picks 1,3,5. The two therefore disagree, which is precisely the + // situation no existing test creates. + // + // With one node per rack, or with racks assigned round-robin, the two would coincide and the test would + // prove nothing. + return super.testClusterConfiguration() + .nodesPerDc(6) + .dcAndRackSupplier((nodeId) -> { + switch (nodeId) + { + case 1: + case 2: + return dcAndRack("datacenter1", "rack1"); + case 3: + case 4: + return dcAndRack("datacenter1", "rack2"); + case 5: + case 6: + return dcAndRack("datacenter1", "rack3"); + default: + return dcAndRack("", ""); + } + }); + } + + @Test + void testReadWithDerivedRangesOnMultiRackCluster() + { + assertCompleteDataset(bulkReaderDataFrame(table).load()); + } + + @Test + void testReadWithCassandraSuppliedRangesOnMultiRackCluster() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load(); + assertCompleteDataset(data); + } + + /** + * The point of the test class: on a rack-aware cluster the two topology sources must still agree on the data + * returned. If they diverge, taking ranges from Cassandra is the only correct option for witness keyspaces. + */ + @Test + void testBothTopologySourcesAgreeOnMultiRackCluster() + { + List derived = sortedNames(bulkReaderDataFrame(table).load()); + List supplied = sortedNames(bulkReaderDataFrame(table, + Collections.singletonMap("forcecassandratokenranges", "true")) + .load()); + assertThat(supplied) + .describedAs("derived and Cassandra-supplied ranges must return the same rows on a rack-aware cluster") + .isEqualTo(derived); + assertThat(derived).hasSize(DATASET.size()); + } + + @Test + void testReadAtLocalQuorumOnMultiRackCluster() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .option("consistencyLevel", "LOCAL_QUORUM") + .load(); + assertCompleteDataset(data); + } + + private void assertCompleteDataset(Dataset data) + { + List rows = data.collectAsList().stream() + .sorted(Comparator.comparing(row -> row.getInt(0))) + .collect(Collectors.toList()); + assertThat(rows).hasSize(DATASET.size()); + for (int i = 0; i < DATASET.size(); i++) + { + assertThat(rows.get(i).getInt(0)).isEqualTo(i); + assertThat(rows.get(i).getString(1)).isEqualTo(DATASET.get(i)); + } + } + + private List sortedNames(Dataset data) + { + return data.collectAsList().stream() + .map(row -> row.getInt(0) + "=" + row.getString(1)) + .sorted() + .collect(Collectors.toList()); + } + + @Override + protected void initializeSchemaForTest() + { + createTestKeyspace(TEST_KEYSPACE, DC1_RF3); + createTestTable(table, "CREATE TABLE IF NOT EXISTS %s (id int PRIMARY KEY, name text);"); + + IInstance firstRunningInstance = cluster.getFirstRunningInstance(); + for (int i = 0; i < DATASET.size(); i++) + { + firstRunningInstance.coordinator() + .execute(String.format("INSERT INTO %s (id, name) VALUES (%d, '%s');", + table, i, DATASET.get(i)), + ConsistencyLevel.ALL); + } + } +} From 5f1b814a0140ac4df52908ef35a3b320b80a0923 Mon Sep 17 00:00:00 2001 From: mkhara Date: Wed, 9 Sep 2026 12:36:11 -0700 Subject: [PATCH 07/11] Reduce the multi-rack test to four nodes Four nodes is the minimum that makes the two topology sources disagree, so the test no longer asks for six. NetworkTopologyStrategy computes acceptableRackRepeats as RF minus rackCount, which is zero with three racks and RF 3, so a node whose rack was already used is skipped: for the range starting at node 1 the reader derives ring order 1,2,3 while Cassandra picks 1,3,4. Six nodes added no coverage and needed two more loopback aliases. --- .../analytics/BulkReaderMultiRackTest.java | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java index 9576dee8e..83b70a856 100644 --- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java @@ -59,17 +59,18 @@ class BulkReaderMultiRackTest extends SharedClusterSparkIntegrationTestBase @Override protected ClusterBuilderConfiguration testClusterConfiguration() { - // Six nodes, three racks, two adjacent nodes per rack, with RF 3. + // Four nodes, three racks, with the first two nodes sharing a rack, and RF 3. // - // The pairing matters. Tokens are handed out in node order, so "the next RF nodes in ring order" - what the - // reader derives - starts 1,2,3. Cassandra cannot use that, because nodes 1 and 2 share rack1, so it skips - // to satisfy one replica per rack and picks 1,3,5. The two therefore disagree, which is precisely the + // NetworkTopologyStrategy computes acceptableRackRepeats = RF - rackCount, which is 0 here, so a node whose + // rack has already been used is skipped. Tokens are handed out in node order, so for the range whose first + // replica is node 1 the reader derives the next three nodes in ring order, 1,2,3, while Cassandra must skip + // node 2 (rack1 again) and pick 1,3,4. The two topology sources therefore genuinely disagree, which is the // situation no existing test creates. // - // With one node per rack, or with racks assigned round-robin, the two would coincide and the test would - // prove nothing. + // Four nodes is the minimum that produces a disagreement. With one node per rack, or with racks assigned + // round-robin, rack-aware selection coincides with ring order and the test would prove nothing. return super.testClusterConfiguration() - .nodesPerDc(6) + .nodesPerDc(4) .dcAndRackSupplier((nodeId) -> { switch (nodeId) { @@ -77,10 +78,8 @@ protected ClusterBuilderConfiguration testClusterConfiguration() case 2: return dcAndRack("datacenter1", "rack1"); case 3: - case 4: return dcAndRack("datacenter1", "rack2"); - case 5: - case 6: + case 4: return dcAndRack("datacenter1", "rack3"); default: return dcAndRack("", ""); From e7d23583f794029bc80ad45877ba8211ed325e86 Mon Sep 17 00:00:00 2001 From: mkhara Date: Wed, 9 Sep 2026 13:51:49 -0700 Subject: [PATCH 08/11] Fail early when Cassandra-reported ranges do not cover the ring A range whose replicas were all skipped, or a topology response that simply omits part of the ring, previously produced a ring with a gap and no error. CassandraRing seeds its range map across the whole ring, so the uncovered range is present but with an empty replica list, and the reader only discovers this once a Spark partition covering it fails with NotEnoughReplicasException - which reads as a cluster availability problem rather than an incomplete topology. Note that TokenPartitioner's coverage assertion does not catch it either. Also deduplicates the replicas recorded per range. PartitionedDataLayer counts replicas per range to decide whether the consistency level is satisfied, so a duplicate would inflate that count and could let a read proceed at a consistency level it has not actually met. --- .../spark/data/CassandraDataLayer.java | 33 +++++++++- .../spark/data/CassandraDataLayerTests.java | 66 +++++++++++++++++++ 2 files changed, 98 insertions(+), 1 deletion(-) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java index 4cb50d252..692f8d138 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java @@ -51,6 +51,8 @@ import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import com.google.common.collect.Range; +import com.google.common.collect.RangeSet; +import com.google.common.collect.TreeRangeSet; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -783,7 +785,9 @@ public CassandraRing createCassandraRingFromTokenRangeReplicas(Partitioner parti } if (!replicas.isEmpty()) { - rangeReplicas.put(range, replicas); + // Distinct, because a duplicated replica would inflate the per-range replica count that + // PartitionedDataLayer uses to decide whether the consistency level is satisfied + rangeReplicas.put(range, replicas.stream().distinct().collect(Collectors.toList())); } } @@ -793,11 +797,38 @@ public CassandraRing createCassandraRingFromTokenRangeReplicas(Partitioner parti "Cassandra reported no read replicas for keyspace " + keyspace + " in datacenter " + datacenter); } + validateCompleteRingCoverage(partitioner, rangeReplicas.keySet()); + LOGGER.info("Using token ranges reported by Cassandra keyspace={} numRanges={} numInstances={}", keyspace, rangeReplicas.size(), instances.size()); return new CassandraRing(partitioner, keyspace, replicationFactor, instances, rangeReplicas); } + /** + * Fails if the supplied ranges leave any part of the token ring without a replica. + *

+ * A gap is not caught downstream. {@link CassandraRing} seeds its range map across the whole ring, so an + * uncovered range is present but with an empty replica list, and the reader only discovers this per Spark + * partition as a {@code NotEnoughReplicasException} that reads like a cluster availability problem. Reporting it + * here names the actual cause. + * + * @param partitioner the partitioner, for the ring bounds + * @param ranges the ranges reported by Cassandra that have at least one usable replica + */ + private void validateCompleteRingCoverage(Partitioner partitioner, Set> ranges) + { + RangeSet uncovered = TreeRangeSet.create(); + uncovered.add(Range.openClosed(partitioner.minToken(), partitioner.maxToken())); + ranges.forEach(uncovered::remove); + if (!uncovered.isEmpty()) + { + throw new IllegalStateException(String.format( + "Token ranges reported by Cassandra do not cover the whole ring for keyspace %s in datacenter %s. " + + "Uncovered: %s. This means some ranges have no replica the reader can use, which would otherwise " + + "surface later as a consistency failure.", keyspace, datacenter, uncovered)); + } + } + // Startup Validation @Override diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java index b69c6eb5b..5fff7a30c 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java @@ -249,4 +249,70 @@ private static List replicaNames(CassandraRing ring, long token) .sorted() .collect(Collectors.toList()); } + + @Test + void testRingFromTokenRangeReplicasFailsWhenRingIsNotFullyCovered() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + // only the lower half of the ring is reported + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "0", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2"); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("do not cover the whole ring"); + } + + @Test + void testRingFromTokenRangeReplicasFailsWhenAllReplicasForARangeAreSkipped() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + // the upper range's only replica is absent from the ring, so that range ends up with no usable + // replica and must be reported as a coverage gap rather than silently dropped + List readReplicas = Arrays.asList( + new ReplicaInfo("-9223372036854775808", "0", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042"))), + new ReplicaInfo("0", "9223372036854775807", + ImmutableMap.of("dc1", Collections.singletonList("1.1.1.9:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2", + "1.1.1.9:9042", "ghost"); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("do not cover the whole ring"); + } + + @Test + void testRingFromTokenRangeReplicasDeduplicatesReplicas() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + // the same replica reported twice must not be counted twice, since PartitionedDataLayer uses the + // per-range replica count to decide whether the consistency level is satisfied + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.1:9042", "1.1.1.2:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology); + + assertThat(result.getReplicas(BigInteger.valueOf(10L))).hasSize(2); + assertThat(replicaNames(result, 10L)).containsExactly("n1", "n2"); + } } From 45d8a2c6b6012d81ff1a1cfcbf762b6b1cf3f821 Mon Sep 17 00:00:00 2001 From: mkhara Date: Wed, 9 Sep 2026 14:04:06 -0700 Subject: [PATCH 09/11] Fix checkstyle violations and clarify what the multi-rack test shows Four formatting violations in the new tests: whitespace after '{' in three array initialisers, and a brace left on the same line as a method signature. Also records what BulkReaderMultiRackTest does and does not establish. Every replica in it is a full replica holding data, so both replica selections return all rows and the reads agree even though Cassandra and the reader choose different replicas. It guards the range mapping and the replica join against regression; it does not show that witness replicas are handled, because choosing the wrong replicas only loses data once some of them hold none. A green run should not be read as covering the witness case. --- .../cassandra/spark/data/ReplicationFactorTests.java | 5 +++-- .../spark/data/partitioner/CassandraRingTests.java | 2 +- .../org/apache/cassandra/spark/utils/CqlUtilsTest.java | 2 +- .../cassandra/analytics/BulkReaderMultiRackTest.java | 8 +++++++- 4 files changed, 12 insertions(+), 5 deletions(-) diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java index 55883b4f7..d1d1ba412 100644 --- a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java +++ b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java @@ -217,7 +217,7 @@ public void testTransientEqualToTotalIsRejected() @Test public void testMalformedTransientValuesAreSkipped() { - for (String malformed : new String[]{ "3/", "/1", "3/1/1", "3/x", "x/1", "3/-1", "" }) + for (String malformed : new String[]{"3/", "/1", "3/1/1", "3/x", "x/1", "3/-1", ""}) { ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( "class", "NetworkTopologyStrategy", @@ -389,7 +389,8 @@ public void testKryoSerializationRoundTripWithTransientReplicas() throws Excepti } @Test - public void testJdkSerializationRoundTripWithTransientReplicas() throws Exception { + public void testJdkSerializationRoundTripWithTransientReplicas() throws Exception + { ReplicationFactor original = new ReplicationFactor(ImmutableMap.of( "class", "NetworkTopologyStrategy", "datacenter1", "3/1", diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java index f31a118ad..1c9e5dbd6 100644 --- a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java +++ b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java @@ -525,7 +525,7 @@ public void testJdkDeserializationRejectsUnknownFormatVersion() throws Exception // Locate the format-version byte deterministically: the format is // [version][partitioner] then writeUTF(keyspace), so it sits two bytes before the "test" UTF header byte[] raw = bytes.toByteArray(); - byte[] keyspaceUtf = new byte[]{ 0, 4, 't', 'e', 's', 't' }; + byte[] keyspaceUtf = new byte[]{0, 4, 't', 'e', 's', 't'}; int keyspaceIndex = indexOf(raw, keyspaceUtf); assertThat(keyspaceIndex).as("keyspace UTF header should be locatable").isGreaterThan(1); int versionIndex = keyspaceIndex - 2; diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java index 3a237a695..ccf44d7e5 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java @@ -197,7 +197,7 @@ public void testExtractReplicationFactorFailsLoudlyOnUnparseableValue() { // An unparseable value must be reported here rather than silently dropping the datacenter, which // would surface later as a confusing "DC not found in replication factor" error - for (String malformed : new String[]{ "xyz", "3/", "3/1/1", "3/x", "3/3" }) + for (String malformed : new String[]{"xyz", "3/", "3/1/1", "3/x", "3/3"}) { String schema = "CREATE KEYSPACE badks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + "'datacenter1': '" + malformed + "'} AND durable_writes = true;\n"; diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java index 83b70a856..6e409a95b 100644 --- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java @@ -48,7 +48,13 @@ * replicas across racks (see {@code acceptableRackRepeats}), so a derived replica set can disagree with Cassandra's. *

* {@code CassandraRing} states that it assumes racks are not in use, so a divergence here would previously have been - * invisible in CI and would surface in production as a bulk read silently returning incomplete data. + * invisible in CI. + *

+ * Note what these tests can and cannot show. Every replica here is a full replica holding data, so both replica + * selections return all the rows and the reads agree. They are therefore a regression guard on the range mapping and + * the replica join, not a demonstration that witness replicas are handled: selecting the wrong replicas only loses + * data once some of them hold none, which requires a mutation tracked keyspace and so cannot be tested until + * Cassandra 6.0 bridge modules land (CASSANALYTICS-192). */ class BulkReaderMultiRackTest extends SharedClusterSparkIntegrationTestBase { From b4880b40f451c4ab81d9957507b04719c7bc9ceb Mon Sep 17 00:00:00 2001 From: mkhara Date: Mon, 14 Sep 2026 10:30:21 -0700 Subject: [PATCH 10/11] CASSANALYTICS-197: Add CHANGES.txt entry --- CHANGES.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGES.txt b/CHANGES.txt index 469bbe82b..dda6945a6 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Source token ranges from Cassandra for mutation tracked keyspaces (CASSANALYTICS-197) * Parse / replication factor for witness-enabled keyspaces (CASSANALYTICS-194) * Determine whether mutation tracking is enabled for keyspace for bulk writes (CASSANALYTICS-160) * Upgrade sidecar version to 0.4.0 From e25b4023f2dffdada35ebb87dc18b6be9f9e0d1e Mon Sep 17 00:00:00 2001 From: mkhara Date: Mon, 14 Sep 2026 11:22:19 -0700 Subject: [PATCH 11/11] CASSANALYTICS-197: Correct the CassandraRing javadoc and align serializer index widths The class javadoc asserted that token ranges are calculated assuming racks are not in use. That is now only true of the local derivation, and it is exactly the assumption the Cassandra-supplied ranges exist to avoid, so it described the opposite of the new behaviour. It now distinguishes the two paths. The JDK serializer wrote replica counts and indexes as shorts while the Kryo serializer used ints. Not a defect, since the instance count is itself a short so indexes cannot exceed its range, but the two paths have to stay in step and the asymmetry would silently truncate if that width ever widened. Both now use int. --- .../spark/data/partitioner/CassandraRing.java | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java index bf525bcf2..d5e461922 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java @@ -58,7 +58,10 @@ * As Cassandra token ranges are dependent on Replication strategy, ring makes sense for a specific keyspace only. * It is made to be immutable for the sake of simplicity. *

- * Token ranges are calculated assuming Cassandra racks are not being used, but controlled by assigning tokens properly. + * Token ranges are either supplied by Cassandra, or calculated locally. The local calculation assumes Cassandra racks + * are not being used, but controlled by assigning tokens properly. Callers that cannot rely on that assumption, such + * as the bulk reader for a mutation tracked keyspace, supply the ranges instead - see + * {@link #CassandraRing(Partitioner, String, ReplicationFactor, Collection, Map)}. *

* {@link #equals(Object)} and {@link #hashCode()} don't take {@link #replicas} and {@link #tokenRangeMap} * into consideration as they are just derived fields. @@ -453,11 +456,11 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE { BigInteger lower = new BigInteger(in.readUTF()); BigInteger upper = new BigInteger(in.readUTF()); - int numReplicas = in.readShort(); + int numReplicas = in.readInt(); List indexes = new ArrayList<>(numReplicas); for (int replica = 0; replica < numReplicas; replica++) { - indexes.add((int) in.readShort()); + indexes.add(in.readInt()); } ranges.add(new ExplicitRange(lower, upper, indexes)); } @@ -509,10 +512,10 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou { out.writeUTF(range.lower.toString()); out.writeUTF(range.upper.toString()); - out.writeShort(range.replicaIndexes.size()); + out.writeInt(range.replicaIndexes.size()); for (int index : range.replicaIndexes) { - out.writeShort(index); + out.writeInt(index); } } }