From 5c783a560d75321905ebd611d39396eec27400ec Mon Sep 17 00:00:00 2001 From: mkhara Date: Fri, 28 Aug 2026 16:46:54 -0700 Subject: [PATCH 1/6] 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 + .../spark/reader/AbstractSchemaBuilder.java | 4 + .../spark/reader/AbstractSchemaBuilder.java | 4 + 11 files changed, 874 insertions(+), 10 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index e62958285..112bf73da 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 * CDC logs NPE for deleted column values (CASSANALYTICS-178) 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 80d69118c..60aa460de 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 d73e1976b..92031dd0f 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 @@ -22,6 +22,7 @@ import java.util.Collections; import java.util.HashMap; +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import org.apache.cassandra.bridge.CassandraBridgeImplementation; @@ -164,4 +165,20 @@ public void testBuildPreservesCdcFlag() .as("build() must preserve enableCdc=false through to the returned CqlTable") .isFalse(); } + + @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/AbstractSchemaBuilder.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java index 52ea0fd82..723c1533a 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java @@ -589,6 +589,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; diff --git a/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java b/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java index aa1c15e8d..a105ebb4b 100644 --- a/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java +++ b/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java @@ -589,6 +589,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 507032810b4e9f2310bfebb462ffb1009c6a55b2 Mon Sep 17 00:00:00 2001 From: mkhara Date: Tue, 15 Sep 2026 13:05:55 -0700 Subject: [PATCH 2/6] CASSANALYTICS-194: Address review - combine replica counts, parse strictly Combines the per-datacenter total and transient counts into a single Map instead of two parallel maps. The maps could previously drift apart, since nothing tied options={dc1=3} to transientOptions={dc1=1}; holding them together makes that unrepresentable. ReplicaCounts becomes a public nested type exposing allReplicas, fullReplicas and transientReplicas. getOptions() and getTransientOptions() are retained, now derived and unmodifiable, so CassandraRing, ConsistencyLevel and PartitionedDataLayer are unaffected. Removes the lenient parse. Its only two callers built hardcoded maps that cannot fail to parse, so it was never needed, and silently dropping a datacenter that Cassandra itself would reject is the wrong default: 3/3 and a negative replication factor now raise rather than being skipped. parseStrict therefore disappears as a separate factory, since strict is the only behaviour. Corrects the javadoc: transient replication predates mutation tracking, and it is witness replicas (CEP-46) that reuse the form. Drops replication_type from the replication factor test fixtures, since Sidecar builds its schema response from the driver's exportAsString() which does not emit that property, and it is irrelevant to replication factor parsing. --- .../spark/data/ReplicationFactor.java | 340 ++++++++++-------- .../cassandra/spark/utils/CqlUtils.java | 6 +- .../spark/data/ReplicationFactorTests.java | 187 ++++------ .../cassandra/spark/utils/CqlUtilsTest.java | 15 +- 4 files changed, 277 insertions(+), 271 deletions(-) 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 ef40bcad8..49ca5191c 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 @@ -19,6 +19,7 @@ package org.apache.cassandra.spark.data; +import java.io.InvalidObjectException; import java.io.Serializable; import java.util.Collections; import java.util.HashMap; @@ -28,8 +29,6 @@ import java.util.Objects; import com.google.common.collect.ImmutableMap; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; @@ -53,15 +52,14 @@ * } *

* 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. + * replicas of which one is transient. Witness replicas (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 { public static final Serializer SERIALIZER = new Serializer(); private static final long serialVersionUID = -2017022813595983257L; - private static final Logger LOGGER = LoggerFactory.getLogger(ReplicationFactor.class); public enum ReplicationStrategy { @@ -112,81 +110,43 @@ public static ReplicationFactor simpleStrategy(int rf) @NotNull 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. + * Per-datacenter replica counts. Holding the total and the transient count together, rather than in two + * parallel maps, means the two cannot drift apart. */ @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); - } + private final Map replication; /** - * 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. + * Parses a raw replication map. A value that cannot be parsed, or that Cassandra itself would reject, raises + * {@link IllegalArgumentException} naming the offending datacenter: a partial replication factor is not usable + * and reporting it here avoids a misleading failure later, such as "DC not found in replication factor". * * @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) + public ReplicationFactor(@NotNull Map options) { this.replicationStrategy = ReplicationFactor.ReplicationStrategy.getEnum(options.get("class")); - this.options = new LinkedHashMap<>(options.size()); - this.transientOptions = new LinkedHashMap<>(); + Map parsed = new LinkedHashMap<>(options.size()); for (Map.Entry entry : options.entrySet()) { if ("class".equals(entry.getKey())) { continue; } - try { - ReplicaCounts counts = ReplicaCounts.parse(entry.getValue()); - this.options.put(entry.getKey(), counts.allReplicas); - if (counts.transientReplicas > 0) - { - this.transientOptions.put(entry.getKey(), counts.transientReplicas); - } + parsed.put(entry.getKey(), ReplicaCounts.parse(entry.getValue())); } 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()); + throw new IllegalArgumentException(String.format("Could not parse replication option: %s = %s", + entry.getKey(), entry.getValue()), exception); } } - - // 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); - } + validateNotEmpty(this.replicationStrategy, parsed, options); + this.replication = Collections.unmodifiableMap(parsed); } public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map options) @@ -197,47 +157,53 @@ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotN 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 (validate && !replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) - { - throw new RuntimeException(String.format("Could not find replication info in schema map: %s.", options)); - } - + Map merged = new LinkedHashMap<>(options.size()); for (Map.Entry entry : options.entrySet()) { if ("class".equals(entry.getKey())) { continue; } - this.options.put(entry.getKey(), entry.getValue()); + int transientReplicas = transientOptions.getOrDefault(entry.getKey(), 0); + merged.put(entry.getKey(), ReplicaCounts.of(entry.getKey(), entry.getValue(), transientReplicas)); } - - for (Map.Entry entry : transientOptions.entrySet()) + for (String datacenter : transientOptions.keySet()) { - if (entry.getValue() == null || entry.getValue() == 0) - { - continue; - } - Integer allReplicas = this.options.get(entry.getKey()); - if (allReplicas == null) + if (!"class".equals(datacenter) && !merged.containsKey(datacenter) + && transientOptions.get(datacenter) != null && transientOptions.get(datacenter) != 0) { throw new IllegalArgumentException(String.format( - "Transient replicas specified for %s but it has no replication factor", entry.getKey())); + "Transient replicas specified for %s but it has no replication factor", datacenter)); } - ReplicaCounts.validate(entry.getKey(), allReplicas, entry.getValue()); - this.transientOptions.put(entry.getKey(), entry.getValue()); + } + validateNotEmpty(replicationStrategy, merged, options); + this.replication = Collections.unmodifiableMap(merged); + } + + /** + * Canonical constructor. {@code validate} exists only for {@link #readResolve()}, which must not reject an + * instance that an earlier version was willing to create. + */ + private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, + @NotNull Map replication, + boolean validate) + { + this.replicationStrategy = replicationStrategy; + if (validate) + { + validateNotEmpty(replicationStrategy, replication, replication); + } + this.replication = Collections.unmodifiableMap(new LinkedHashMap<>(replication)); + } + + private static void validateNotEmpty(ReplicationStrategy strategy, Map parsed, Object raw) + { + // A strategy other than LocalStrategy with no datacenter entries is not usable + if (strategy != ReplicationStrategy.LocalStrategy && parsed.isEmpty()) + { + throw new IllegalArgumentException("Could not find replication info in schema map: " + raw); } } @@ -247,9 +213,7 @@ private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, */ public Integer getTotalReplicationFactor() { - return options.values().stream() - .mapToInt(Integer::intValue) - .sum(); + return replication.values().stream().mapToInt(ReplicaCounts::allReplicas).sum(); } /** @@ -258,7 +222,7 @@ public Integer getTotalReplicationFactor() */ public Integer getFullReplicationFactor() { - return getTotalReplicationFactor() - getTransientReplicationFactor(); + return replication.values().stream().mapToInt(ReplicaCounts::fullReplicas).sum(); } /** @@ -266,9 +230,7 @@ public Integer getFullReplicationFactor() */ public Integer getTransientReplicationFactor() { - return transientOptions.values().stream() - .mapToInt(Integer::intValue) - .sum(); + return replication.values().stream().mapToInt(ReplicaCounts::transientReplicas).sum(); } /** @@ -276,7 +238,7 @@ public Integer getTransientReplicationFactor() */ public boolean hasTransientReplicas() { - return !transientOptions.isEmpty(); + return replication.values().stream().anyMatch(counts -> counts.transientReplicas() > 0); } /** @@ -285,7 +247,8 @@ public boolean hasTransientReplicas() */ public int getTransientReplicas(@NotNull String datacenter) { - return transientOptions.getOrDefault(datacenter, 0); + ReplicaCounts counts = replication.get(datacenter); + return counts == null ? 0 : counts.transientReplicas(); } /** @@ -295,31 +258,56 @@ public int getTransientReplicas(@NotNull String datacenter) */ public int getFullReplicas(@NotNull String datacenter) { - Integer allReplicas = options.get(datacenter); - if (allReplicas == null) + return counts(datacenter).fullReplicas(); + } + + private ReplicaCounts counts(@NotNull String datacenter) + { + ReplicaCounts counts = replication.get(datacenter); + if (counts == null) { throw new IllegalArgumentException(String.format("Datacenter %s not found in replication factor %s", - datacenter, options.keySet())); + datacenter, replication.keySet())); } - return allReplicas - getTransientReplicas(datacenter); + return counts; } /** - * @return per-datacenter total replica counts, including transient (witness) replicas + * @return per-datacenter replica counts, unmodifiable + */ + @NotNull + public Map getReplication() + { + return replication; + } + + /** + * @return per-datacenter total replica counts, including transient (witness) replicas. Unmodifiable, and derived + * from {@link #getReplication()}. */ @NotNull public Map getOptions() { - return options; + Map options = new LinkedHashMap<>(replication.size()); + replication.forEach((datacenter, counts) -> options.put(datacenter, counts.allReplicas())); + return Collections.unmodifiableMap(options); } /** - * @return per-datacenter transient (witness) replica counts. Datacenters without transient replicas are absent. + * @return per-datacenter transient (witness) replica counts, unmodifiable. Datacenters without transient + * replicas are absent. */ @NotNull public Map getTransientOptions() { - return transientOptions; + Map transientOptions = new LinkedHashMap<>(); + replication.forEach((datacenter, counts) -> { + if (counts.transientReplicas() > 0) + { + transientOptions.put(datacenter, counts.transientReplicas()); + } + }); + return Collections.unmodifiableMap(transientOptions); } @NotNull @@ -346,42 +334,42 @@ 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.transientOptions, that.transientOptions); + && Objects.equals(this.replication, that.replication); } @Override public int hashCode() { - return Objects.hash(replicationStrategy, options, transientOptions); + return Objects.hash(replicationStrategy, replication); } /** - * {@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. + * {@link #serialVersionUID} is pinned, so an instance serialized before the per-datacenter counts were combined + * deserializes with a {@code null} {@code replication}. Only reachable under driver/executor version skew, which + * is not a supported configuration, so this reports the mismatch rather than silently losing the counts. * - * @return this instance, or a normalised copy when deserialized from the older form + * @return this instance + * @throws InvalidObjectException when deserialized from an incompatible older form */ - private Object readResolve() + private Object readResolve() throws InvalidObjectException { - if (transientOptions != null) + if (replication == null) { - return this; + throw new InvalidObjectException( + "ReplicationFactor was serialized by an incompatible version: per-datacenter replica counts are absent. " + + "Driver and executors must run the same version."); } - // 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); + return this; } /** - * 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. + * Replica counts for a single datacenter. Cassandra accepts either {@code } or + * {@code /}; transient replication predates mutation tracking, and witness replicas + * (CEP-46) reuse the same form. See {@code org.apache.cassandra.locator.ReplicationFactor} in Cassandra. */ - private static final class ReplicaCounts + public static final class ReplicaCounts implements Serializable { + private static final long serialVersionUID = 2026091500000000001L; private static final String TRANSIENT_SEPARATOR = "/"; private final int allReplicas; @@ -393,6 +381,43 @@ private ReplicaCounts(int allReplicas, int transientReplicas) this.transientReplicas = transientReplicas; } + /** + * @param datacenter datacenter name, used only in error messages, may be {@code null} + * @param allReplicas total replicas + * @param transientReplicas transient (witness) replicas + * @return validated counts + * @throws IllegalArgumentException when the counts are inconsistent + */ + public static ReplicaCounts of(String datacenter, int allReplicas, int transientReplicas) + { + validate(datacenter, allReplicas, transientReplicas); + return new ReplicaCounts(allReplicas, transientReplicas); + } + + /** + * @return total replicas, including transient (witness) replicas + */ + public int allReplicas() + { + return allReplicas; + } + + /** + * @return replicas holding the full data set + */ + public int fullReplicas() + { + return allReplicas - transientReplicas; + } + + /** + * @return transient (witness) replicas, {@code 0} when none are configured + */ + public int transientReplicas() + { + return transientReplicas; + } + /** * @param value the raw replication value, e.g. {@code "3"} or {@code "3/1"} * @return the parsed replica counts @@ -405,9 +430,7 @@ static ReplicaCounts parse(@NotNull String value) int separator = trimmed.indexOf(TRANSIENT_SEPARATOR); if (separator < 0) { - int allReplicas = Integer.parseInt(trimmed); - validate(null, allReplicas, 0); - return new ReplicaCounts(allReplicas, 0); + return of(null, Integer.parseInt(trimmed), 0); } if (trimmed.indexOf(TRANSIENT_SEPARATOR, separator + 1) >= 0) @@ -416,10 +439,9 @@ static ReplicaCounts parse(@NotNull String value) "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); + return of(null, + Integer.parseInt(trimmed.substring(0, separator).trim()), + Integer.parseInt(trimmed.substring(separator + 1).trim())); } /** @@ -450,6 +472,34 @@ static void validate(String datacenter, int allReplicas, int transientReplicas) allReplicas, transientReplicas, where)); } } + + @Override + public boolean equals(Object other) + { + if (this == other) + { + return true; + } + if (other == null || getClass() != other.getClass()) + { + return false; + } + ReplicaCounts that = (ReplicaCounts) other; + return allReplicas == that.allReplicas && transientReplicas == that.transientReplicas; + } + + @Override + public int hashCode() + { + return Objects.hash(allReplicas, transientReplicas); + } + + @Override + public String toString() + { + return transientReplicas > 0 ? allReplicas + TRANSIENT_SEPARATOR + transientReplicas + : String.valueOf(allReplicas); + } } public static class Serializer extends com.esotericsoftware.kryo.Serializer @@ -458,19 +508,12 @@ public static class Serializer extends com.esotericsoftware.kryo.Serializer entry : replicationFactor.options.entrySet()) - { - 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.writeByte(replicationFactor.replication.size()); + for (Map.Entry entry : replicationFactor.replication.entrySet()) { out.writeString(entry.getKey()); - out.writeByte(entry.getValue()); + out.writeByte(entry.getValue().allReplicas()); + out.writeByte(entry.getValue().transientReplicas()); } } @@ -478,19 +521,18 @@ public void write(Kryo kryo, Output out, ReplicationFactor replicationFactor) public ReplicationFactor read(Kryo kryo, Input in, Class type) { ReplicationStrategy strategy = ReplicationStrategy.valueOf(in.readByte()); - int numOptions = in.readByte(); - Map options = new HashMap<>(numOptions); - for (int option = 0; option < numOptions; option++) - { - options.put(in.readString(), (int) in.readByte()); - } - int numTransientOptions = in.readByte(); - Map transientOptions = new HashMap<>(numTransientOptions); - for (int option = 0; option < numTransientOptions; option++) + int numDatacenters = in.readByte(); + Map replication = new LinkedHashMap<>(numDatacenters); + for (int datacenter = 0; datacenter < numDatacenters; datacenter++) { - transientOptions.put(in.readString(), (int) in.readByte()); + String name = in.readString(); + int allReplicas = in.readByte(); + int transientReplicas = in.readByte(); + replication.put(name, ReplicaCounts.of(name, allReplicas, transientReplicas)); } - return new ReplicationFactor(strategy, options, transientOptions); + // Validation is skipped: an instance that was serialized was already valid, and LocalStrategy + // legitimately has no datacenter entries + return new ReplicationFactor(strategy, replication, false); } } } 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 60aa460de..59dc919b8 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 @@ -177,11 +177,11 @@ public static ReplicationFactor extractReplicationFactor(@NotNull String schemaS } // 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. + // ReplicationFactor, which reports an unparseable value rather than dropping the datacenter. Dropping it + // would surface later as a confusing "DC not found in replication factor" error. try { - return ReplicationFactor.parseStrict(map); + return new ReplicationFactor(map); } catch (IllegalArgumentException exception) { 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..bfc528972 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 @@ -207,24 +207,24 @@ public void testTransientReplicasWithWhitespace() @Test public void testTransientEqualToTotalIsRejected() { - // Cassandra requires at least one full replica, so 3/3 is invalid - ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + // Cassandra requires at least one full replica, so 3/3 is invalid and must not be accepted silently + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( "class", "NetworkTopologyStrategy", - "datacenter1", "3/3")); - assertThat(replicationFactor.getOptions()).doesNotContainKey("datacenter1"); + "datacenter1", "3/3"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("datacenter1"); } @Test - public void testMalformedTransientValuesAreSkipped() + public void testMalformedReplicationValuesAreRejected() { - 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( + assertThatThrownBy(() -> 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"); + "datacenter1", malformed))) + .as("malformed replication value '%s' should be rejected", malformed) + .isInstanceOf(IllegalArgumentException.class); } } @@ -264,10 +264,11 @@ public void testEqualityConsidersTransientReplicas() @Test public void testNegativeReplicationFactorIsRejected() { - ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( "class", "NetworkTopologyStrategy", - "datacenter1", "-3")); - assertThat(replicationFactor.getOptions()).doesNotContainKey("datacenter1"); + "datacenter1", "-3"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("datacenter1"); } @Test @@ -284,81 +285,6 @@ public void testZeroReplicationFactorIsAllowed() // 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 { @@ -412,31 +338,70 @@ public void testJdkSerializationRoundTripWithTransientReplicas() throws Exceptio } @Test - public void testReadResolveToleratesLegacyEmptyOptions() throws Exception + public void testUnparseableValueIsRejected() { - // 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(); + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "xyz"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("datacenter1"); + } - Field transientOptions = ReplicationFactor.class.getDeclaredField("transientOptions"); - transientOptions.setAccessible(true); - transientOptions.set(legacy, null); + @Test + public void testNoDatacenterEntriesIsRejected() + { + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Could not find replication info"); + } - 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(); - } + @Test + public void testLocalStrategyWithNoEntriesIsAllowed() + { + // LocalStrategy legitimately has no datacenter entries, e.g. the system_schema keyspace + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.LocalStrategy")); + assertThat(replicationFactor.getReplicationStrategy()) + .isEqualTo(ReplicationFactor.ReplicationStrategy.LocalStrategy); + assertThat(replicationFactor.getOptions()).isEmpty(); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(0); + } - assertThat(deserialized.getOptions()).isEmpty(); - assertThat(deserialized.hasTransientReplicas()).isFalse(); - assertThat(deserialized.getTransientReplicationFactor()).isEqualTo(0); + // The per-datacenter counts are held together rather than in two parallel maps, so the total and the + // transient count cannot drift apart + + @Test + public void testReplicationExposesCombinedCounts() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + + assertThat(replicationFactor.getReplication()).containsOnlyKeys("datacenter1", "datacenter2"); + ReplicationFactor.ReplicaCounts dc1 = replicationFactor.getReplication().get("datacenter1"); + assertThat(dc1.allReplicas()).isEqualTo(3); + assertThat(dc1.fullReplicas()).isEqualTo(2); + assertThat(dc1.transientReplicas()).isEqualTo(1); + + ReplicationFactor.ReplicaCounts dc2 = replicationFactor.getReplication().get("datacenter2"); + assertThat(dc2.allReplicas()).isEqualTo(3); + assertThat(dc2.fullReplicas()).isEqualTo(3); + assertThat(dc2.transientReplicas()).isEqualTo(0); + } + + @Test + public void testDerivedOptionMapsAreUnmodifiable() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThatThrownBy(() -> replicationFactor.getOptions().put("datacenter2", 3)) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> replicationFactor.getTransientOptions().put("datacenter2", 1)) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> replicationFactor.getReplication().remove("datacenter1")) + .isInstanceOf(UnsupportedOperationException.class); } } 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..c06c869f8 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 @@ -145,11 +145,12 @@ public void testExtractReplicationFactor(CassandraBridge bridge) @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' + // The / form, as Cassandra's WitnessAlwaysReadsFullReplicaTest creates it on the + // cep-45-mutation-tracking branch. replication_type is deliberately omitted: Sidecar builds its schema + // response from the driver's exportAsString(), which does not emit that property, and it is irrelevant to + // replication factor parsing in any case. extractReplicationType is covered separately. String schema = "CREATE KEYSPACE witnessks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " - + "'datacenter1': '3/1', 'datacenter2': '3/1'} AND replication_type = 'tracked' " - + "AND durable_writes = true;\n"; + + "'datacenter1': '3/1', 'datacenter2': '3/1'} AND durable_writes = true;\n"; ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "witnessks"); assertThat(rf).isNotNull(); @@ -162,15 +163,13 @@ public void testExtractReplicationFactorWithWitnessReplicas() 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"; + + "'datacenter1': '3/1', 'datacenter2': '3'} AND durable_writes = true;\n"; ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "mixedks"); assertThat(rf.getTotalReplicationFactor()).isEqualTo(6); @@ -197,7 +196,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"; From 7646f426d796d5f565f729b6b308f539e9b5f35c Mon Sep 17 00:00:00 2001 From: mkhara Date: Tue, 15 Sep 2026 13:46:41 -0700 Subject: [PATCH 3/6] CASSANALYTICS-194: Remove stale comment referencing the deleted parseStrict factory --- .../org/apache/cassandra/spark/data/ReplicationFactorTests.java | 2 -- 1 file changed, 2 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 bfc528972..b82dd1509 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 @@ -283,8 +283,6 @@ public void testZeroReplicationFactorIsAllowed() assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); } - // parseStrict: same parsing, but an unparseable value raises instead of dropping the datacenter - @Test public void testKryoSerializationRoundTripWithTransientReplicas() throws Exception { From fab1b7d3890fb7e0cd513a194b6dac8e49078f50 Mon Sep 17 00:00:00 2001 From: mkhara Date: Wed, 16 Sep 2026 18:25:49 -0700 Subject: [PATCH 4/6] CASSANALYTICS-194: Address second review round Restructures the constructors so each one resolves its input to per-datacenter counts and delegates to a single canonical constructor, rather than repeating the class lookup, the class-key skip and the empty check. Parsing raw strings and merging separate total and transient maps move into named helpers. Removes the validate flag from the private constructor. It was only ever passed false, so the guarded branch was dead: the flag was left over from an earlier version of readResolve that reconstructed the instance instead of rejecting it. CassandraRing now writes the ReplicationFactor as an object rather than destructuring it into strategy, options and transient options. ReplicationFactor is Serializable, so this is both shorter and independent of how it represents per-datacenter counts, which is what made the previous version need updating in the first place. Drops the explanatory comment from the replication factor test fixture. --- .../spark/data/ReplicationFactor.java | 117 +++++++++--------- .../spark/data/partitioner/CassandraRing.java | 37 +----- .../cassandra/spark/utils/CqlUtilsTest.java | 4 - 3 files changed, 66 insertions(+), 92 deletions(-) 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 49ca5191c..9974b4aa3 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 @@ -127,26 +127,7 @@ public static ReplicationFactor simpleStrategy(int rf) */ public ReplicationFactor(@NotNull Map options) { - this.replicationStrategy = ReplicationFactor.ReplicationStrategy.getEnum(options.get("class")); - Map parsed = new LinkedHashMap<>(options.size()); - for (Map.Entry entry : options.entrySet()) - { - if ("class".equals(entry.getKey())) - { - continue; - } - try - { - parsed.put(entry.getKey(), ReplicaCounts.parse(entry.getValue())); - } - catch (IllegalArgumentException exception) - { - throw new IllegalArgumentException(String.format("Could not parse replication option: %s = %s", - entry.getKey(), entry.getValue()), exception); - } - } - validateNotEmpty(this.replicationStrategy, parsed, options); - this.replication = Collections.unmodifiableMap(parsed); + this(ReplicationStrategy.getEnum(options.get("class")), parse(options), options); } public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map options) @@ -158,53 +139,77 @@ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map options, @NotNull Map transientOptions) { - this.replicationStrategy = replicationStrategy; - Map merged = new LinkedHashMap<>(options.size()); - for (Map.Entry entry : options.entrySet()) - { - if ("class".equals(entry.getKey())) - { - continue; - } - int transientReplicas = transientOptions.getOrDefault(entry.getKey(), 0); - merged.put(entry.getKey(), ReplicaCounts.of(entry.getKey(), entry.getValue(), transientReplicas)); - } - for (String datacenter : transientOptions.keySet()) - { - if (!"class".equals(datacenter) && !merged.containsKey(datacenter) - && transientOptions.get(datacenter) != null && transientOptions.get(datacenter) != 0) - { - throw new IllegalArgumentException(String.format( - "Transient replicas specified for %s but it has no replication factor", datacenter)); - } - } - validateNotEmpty(replicationStrategy, merged, options); - this.replication = Collections.unmodifiableMap(merged); + this(replicationStrategy, merge(options, transientOptions), options); } /** - * Canonical constructor. {@code validate} exists only for {@link #readResolve()}, which must not reject an - * instance that an earlier version was willing to create. + * Canonical constructor. Every other constructor resolves its input to per-datacenter counts and delegates here. + * + * @param replicationStrategy the replication strategy + * @param replication per-datacenter replica counts, already validated individually + * @param raw the caller's original input, used only in the error message */ private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map replication, - boolean validate) + @NotNull Object raw) { - this.replicationStrategy = replicationStrategy; - if (validate) + // A strategy other than LocalStrategy with no datacenter entries is not usable + if (replicationStrategy != ReplicationStrategy.LocalStrategy && replication.isEmpty()) { - validateNotEmpty(replicationStrategy, replication, replication); + throw new IllegalArgumentException("Could not find replication info in schema map: " + raw); } + this.replicationStrategy = replicationStrategy; this.replication = Collections.unmodifiableMap(new LinkedHashMap<>(replication)); } - private static void validateNotEmpty(ReplicationStrategy strategy, Map parsed, Object raw) + /** + * Resolves raw string values, e.g. {@code "3"} or {@code "3/1"}, to per-datacenter counts. + */ + private static Map parse(@NotNull Map options) + { + Map parsed = new LinkedHashMap<>(options.size()); + options.forEach((datacenter, value) -> { + if ("class".equals(datacenter)) + { + return; + } + try + { + parsed.put(datacenter, ReplicaCounts.parse(value)); + } + catch (IllegalArgumentException exception) + { + throw new IllegalArgumentException(String.format("Could not parse replication option: %s = %s", + datacenter, value), exception); + } + }); + return parsed; + } + + /** + * Resolves separate total and transient maps to per-datacenter counts, rejecting a transient entry for a + * datacenter that has no replication factor. + */ + private static Map merge(@NotNull Map options, + @NotNull Map transientOptions) { - // A strategy other than LocalStrategy with no datacenter entries is not usable - if (strategy != ReplicationStrategy.LocalStrategy && parsed.isEmpty()) - { - throw new IllegalArgumentException("Could not find replication info in schema map: " + raw); - } + Map merged = new LinkedHashMap<>(options.size()); + options.forEach((datacenter, allReplicas) -> { + if (!"class".equals(datacenter)) + { + merged.put(datacenter, + ReplicaCounts.of(datacenter, allReplicas, transientOptions.getOrDefault(datacenter, 0))); + } + }); + transientOptions.forEach((datacenter, transientReplicas) -> { + if (!"class".equals(datacenter) && transientReplicas != null && transientReplicas != 0 + && !merged.containsKey(datacenter)) + { + throw new IllegalArgumentException(String.format( + "Transient replicas specified for %s but it has no replication factor", datacenter)); + } + }); + return merged; } /** @@ -530,9 +535,7 @@ public ReplicationFactor read(Kryo kryo, Input in, Class type int transientReplicas = in.readByte(); replication.put(name, ReplicaCounts.of(name, allReplicas, transientReplicas)); } - // Validation is skipped: an instance that was serialized was already valid, and LocalStrategy - // legitimately has no datacenter entries - return new ReplicationFactor(strategy, replication, false); + return new ReplicationFactor(strategy, replication, replication); } } } 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..096a5bf14 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 @@ -78,8 +78,8 @@ public class CassandraRing implements Serializable 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. + * Incremented whenever the hand-rolled JDK serialization format below changes. Version 1 writes the + * ReplicationFactor as an object rather than destructuring it. */ private static final byte SERIALIZATION_FORMAT_VERSION = 1; @@ -283,20 +283,9 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE this.partitioner = in.readByte() == 0 ? Partitioner.RandomPartitioner : Partitioner.Murmur3Partitioner; this.keyspace = in.readUTF(); - ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.valueOf(in.readByte()); - int optionCount = in.readByte(); - Map options = new HashMap<>(optionCount); - for (int option = 0; option < optionCount; option++) - { - options.put(in.readUTF(), (int) in.readByte()); - } - 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); + // ReplicationFactor is Serializable, so it is written whole rather than destructured. That keeps this + // method independent of how ReplicationFactor represents its per-datacenter counts. + this.replicationFactor = (ReplicationFactor) in.readObject(); int numInstances = in.readShort(); this.instances = new ArrayList<>(numInstances); @@ -314,21 +303,7 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou out.writeByte(this.partitioner == Partitioner.RandomPartitioner ? 0 : 1); out.writeUTF(this.keyspace); - out.writeByte(this.replicationFactor.getReplicationStrategy().value); - Map options = this.replicationFactor.getOptions(); - out.writeByte(options.size()); - for (Map.Entry option : options.entrySet()) - { - 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.writeObject(this.replicationFactor); out.writeShort(this.instances.size()); for (CassandraInstance instance : this.instances) 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 c06c869f8..c2cf777a8 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 @@ -145,10 +145,6 @@ public void testExtractReplicationFactor(CassandraBridge bridge) @Test public void testExtractReplicationFactorWithWitnessReplicas() { - // The / form, as Cassandra's WitnessAlwaysReadsFullReplicaTest creates it on the - // cep-45-mutation-tracking branch. replication_type is deliberately omitted: Sidecar builds its schema - // response from the driver's exportAsString(), which does not emit that property, and it is irrelevant to - // replication factor parsing in any case. extractReplicationType is covered separately. String schema = "CREATE KEYSPACE witnessks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + "'datacenter1': '3/1', 'datacenter2': '3/1'} AND durable_writes = true;\n"; From c5c715c39072b9f8f387a073f5b03543f893d13a Mon Sep 17 00:00:00 2001 From: mkhara Date: Thu, 17 Sep 2026 10:46:53 -0700 Subject: [PATCH 5/6] CASSANALYTICS-194: Precompute derived counts, and reject unknown datacenter consistently The aggregate counts and the two derived per-datacenter maps were recomputed on every call, and getOptions in particular is called in a loop from CassandraRing.init. They are now computed once in the canonical constructor, which this class can do because it is immutable. The derived fields are transient, so they are absent from the serialized form rather than duplicating what replication already holds. readResolve rebuilds through the canonical constructor, and the Kryo serializer already went through it, so neither path can leave them out of step. Tests cover both round trips. getTransientReplicas returned 0 for a datacenter with no replication factor while getFullReplicas threw for the same input. That conflated "no transient replicas" with "not a datacenter of this keyspace" and hid the second case. It now throws, matching getFullReplicas. No production caller passes an unknown datacenter. --- .../spark/data/ReplicationFactor.java | 60 +++++++++++----- .../spark/data/ReplicationFactorTests.java | 69 +++++++++++++++++++ 2 files changed, 110 insertions(+), 19 deletions(-) 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 9974b4aa3..229c06972 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 @@ -117,6 +117,15 @@ public static ReplicationFactor simpleStrategy(int rf) @NotNull private final Map replication; + // Derived from replication, computed once because this class is immutable. Marked transient so they are absent + // from the serialized form: readResolve rebuilds through the canonical constructor, and the Kryo serializer + // already goes through it, so neither path can leave these out of step with replication. + private final transient int totalReplicationFactor; + private final transient int fullReplicationFactor; + private final transient int transientReplicationFactor; + private final transient Map options; + private final transient Map transientOptions; + /** * Parses a raw replication map. A value that cannot be parsed, or that Cassandra itself would reject, raises * {@link IllegalArgumentException} naming the offending datacenter: a partial replication factor is not usable @@ -160,6 +169,27 @@ private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, } this.replicationStrategy = replicationStrategy; this.replication = Collections.unmodifiableMap(new LinkedHashMap<>(replication)); + + int total = 0; + int transientCount = 0; + Map allByDatacenter = new LinkedHashMap<>(this.replication.size()); + Map transientByDatacenter = new LinkedHashMap<>(); + for (Map.Entry entry : this.replication.entrySet()) + { + ReplicaCounts counts = entry.getValue(); + total += counts.allReplicas(); + transientCount += counts.transientReplicas(); + allByDatacenter.put(entry.getKey(), counts.allReplicas()); + if (counts.transientReplicas() > 0) + { + transientByDatacenter.put(entry.getKey(), counts.transientReplicas()); + } + } + this.totalReplicationFactor = total; + this.transientReplicationFactor = transientCount; + this.fullReplicationFactor = total - transientCount; + this.options = Collections.unmodifiableMap(allByDatacenter); + this.transientOptions = Collections.unmodifiableMap(transientByDatacenter); } /** @@ -218,7 +248,7 @@ private static Map merge(@NotNull Map op */ public Integer getTotalReplicationFactor() { - return replication.values().stream().mapToInt(ReplicaCounts::allReplicas).sum(); + return totalReplicationFactor; } /** @@ -227,7 +257,7 @@ public Integer getTotalReplicationFactor() */ public Integer getFullReplicationFactor() { - return replication.values().stream().mapToInt(ReplicaCounts::fullReplicas).sum(); + return fullReplicationFactor; } /** @@ -235,7 +265,7 @@ public Integer getFullReplicationFactor() */ public Integer getTransientReplicationFactor() { - return replication.values().stream().mapToInt(ReplicaCounts::transientReplicas).sum(); + return transientReplicationFactor; } /** @@ -243,17 +273,17 @@ public Integer getTransientReplicationFactor() */ public boolean hasTransientReplicas() { - return replication.values().stream().anyMatch(counts -> counts.transientReplicas() > 0); + 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 + * @return the number of transient (witness) replicas in {@code datacenter}, {@code 0} when it has none + * @throws IllegalArgumentException when {@code datacenter} has no replication factor */ public int getTransientReplicas(@NotNull String datacenter) { - ReplicaCounts counts = replication.get(datacenter); - return counts == null ? 0 : counts.transientReplicas(); + return counts(datacenter).transientReplicas(); } /** @@ -293,9 +323,7 @@ public Map getReplication() @NotNull public Map getOptions() { - Map options = new LinkedHashMap<>(replication.size()); - replication.forEach((datacenter, counts) -> options.put(datacenter, counts.allReplicas())); - return Collections.unmodifiableMap(options); + return options; } /** @@ -305,14 +333,7 @@ public Map getOptions() @NotNull public Map getTransientOptions() { - Map transientOptions = new LinkedHashMap<>(); - replication.forEach((datacenter, counts) -> { - if (counts.transientReplicas() > 0) - { - transientOptions.put(datacenter, counts.transientReplicas()); - } - }); - return Collections.unmodifiableMap(transientOptions); + return transientOptions; } @NotNull @@ -364,7 +385,8 @@ private Object readResolve() throws InvalidObjectException "ReplicationFactor was serialized by an incompatible version: per-datacenter replica counts are absent. " + "Driver and executors must run the same version."); } - return this; + // Rebuild through the canonical constructor so the derived fields, which are transient, are populated + return new ReplicationFactor(replicationStrategy, replication, replication); } /** 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 b82dd1509..aa0f5f384 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 @@ -402,4 +402,73 @@ public void testDerivedOptionMapsAreUnmodifiable() assertThatThrownBy(() -> replicationFactor.getReplication().remove("datacenter1")) .isInstanceOf(UnsupportedOperationException.class); } + + @Test + public void testGetTransientReplicasUnknownDatacenter() + { + // Consistent with getFullReplicas: an unknown datacenter is a programming error, not zero transients + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThatThrownBy(() -> replicationFactor.getTransientReplicas("nosuchdc")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("nosuchdc"); + } + + @Test + public void testPrecomputedValuesSurviveJdkSerialization() throws Exception + { + // The aggregate counts and derived maps are precomputed and transient, so they are absent from the + // serialized form and must be rebuilt on the way back in + 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.getTotalReplicationFactor()).isEqualTo(6); + assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5); + assertThat(deserialized.getTransientReplicationFactor()).isEqualTo(1); + assertThat(deserialized.hasTransientReplicas()).isTrue(); + assertThat(deserialized.getOptions()).containsOnlyKeys("datacenter1", "datacenter2"); + assertThat(deserialized.getTransientOptions()).containsOnlyKeys("datacenter1"); + assertThat(deserialized).isEqualTo(original); + } + + @Test + public void testPrecomputedValuesSurviveKryoSerialization() + { + 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.getTotalReplicationFactor()).isEqualTo(6); + assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5); + assertThat(deserialized.getOptions()).containsOnlyKeys("datacenter1", "datacenter2"); + assertThat(deserialized.getTransientOptions()).containsOnlyKeys("datacenter1"); + } } From e3733e49a784ea97102070f45f9b2e90c75c9aa2 Mon Sep 17 00:00:00 2001 From: mkhara Date: Mon, 21 Sep 2026 13:01:49 -0700 Subject: [PATCH 6/6] CASSANALYTICS-194: Add rfToMap transient-replica test to the six-zero bridge Cassandra 6.0 support added cassandra-six-zero-bridge after this branch was written. Mirror the four-zero and five-zero coverage so the bridge that witness keyspaces actually run on asserts the same behaviour. --- .../spark/reader/SchemaBuilderTests.java | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java b/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java index f378982c7..ba7896712 100644 --- a/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java +++ b/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java @@ -138,4 +138,20 @@ public boolean compatibleWith(ClusterMetadata metadata) 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"); + } }