This is an automated email from the ASF dual-hosted git repository. sarankk pushed a commit to branch mutation-tracking-support in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
commit 9cf7d02679138baa6aa1ef688571d2ecf6e067c9 Author: Mansi Khara <[email protected]> AuthorDate: Thu Oct 1 19:46:06 2026 -0400 CASSANALYTICS-194: Parse <replicas>/<transient> replication factor for witness-enabled keyspaces (#238) Patch by Mansi Khara; Reviewed by Saranya Krishnakumar, Jyothsna Konisa, Bernardo Botella for CASSANALYTICS-194 --- CHANGES.txt | 1 + .../cassandra/spark/data/ReplicationFactor.java | 462 +++++++++++++++++++-- .../spark/data/partitioner/CassandraRing.java | 43 +- .../org/apache/cassandra/spark/utils/CqlUtils.java | 15 +- .../spark/data/ReplicationFactorTests.java | 338 +++++++++++++++ .../spark/data/partitioner/CassandraRingTests.java | 104 +++++ .../apache/cassandra/spark/utils/CqlUtilsTest.java | 73 ++++ .../cassandra/spark/reader/SchemaBuilderTests.java | 16 + .../cassandra/spark/reader/SchemaBuilderTests.java | 17 + .../spark/reader/AbstractSchemaBuilder.java | 4 + .../cassandra/spark/reader/SchemaBuilderTests.java | 16 + .../spark/reader/AbstractSchemaBuilder.java | 4 + 12 files changed, 1033 insertions(+), 60 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 95fe46d6..dbb7415e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Parse <replicas>/<transient> 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 * Opening one keyspace instance rebuilds every keyspace instance in the Cassandra 6.0 bridge (CASSANALYTICS-202) 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 0b9b6821..9abc88ad 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,16 +19,15 @@ package org.apache.cassandra.spark.data; +import java.io.InvalidObjectException; import java.io.Serializable; -import java.util.HashMap; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; 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; @@ -50,12 +49,16 @@ import org.jetbrains.annotations.NotNull; * "replication_factor" : 1 * } * } + * <p> + * Replica counts may also use the {@code <replicas>/<transient>} form, e.g. {@code "DC1" : "3/1"}, meaning three + * 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 { @@ -106,64 +109,257 @@ public class ReplicationFactor implements Serializable @NotNull private final ReplicationStrategy replicationStrategy; + /** + * 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<String, Integer> options; + private final Map<String, ReplicaCounts> 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<String, Integer> options; + private final transient Map<String, Integer> 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 + * 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 + * @throws IllegalArgumentException when any replication value cannot be parsed + */ public ReplicationFactor(@NotNull Map<String, String> options) { - this.replicationStrategy = ReplicationFactor.ReplicationStrategy.getEnum(options.get("class")); - this.options = new LinkedHashMap<>(options.size()); - for (Map.Entry<String, String> entry : options.entrySet()) + this(ReplicationStrategy.getEnum(options.get("class")), parse(options), options); + } + + public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map<String, Integer> options) + { + this(replicationStrategy, options, Collections.emptyMap()); + } + + public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, + @NotNull Map<String, Integer> options, + @NotNull Map<String, Integer> transientOptions) + { + this(replicationStrategy, merge(options, transientOptions), options); + } + + /** + * 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<String, ReplicaCounts> replication, + @NotNull Object raw) + { + // A strategy other than LocalStrategy with no datacenter entries is not usable + if (replicationStrategy != ReplicationStrategy.LocalStrategy && replication.isEmpty()) + { + throw new IllegalArgumentException("Could not find replication info in schema map: " + raw); + } + this.replicationStrategy = replicationStrategy; + this.replication = Collections.unmodifiableMap(new LinkedHashMap<>(replication)); + + int total = 0; + int transientCount = 0; + Map<String, Integer> allByDatacenter = new LinkedHashMap<>(this.replication.size()); + Map<String, Integer> transientByDatacenter = new LinkedHashMap<>(); + for (Map.Entry<String, ReplicaCounts> entry : this.replication.entrySet()) { - if ("class".equals(entry.getKey())) + ReplicaCounts counts = entry.getValue(); + total += counts.allReplicas(); + transientCount += counts.transientReplicas(); + allByDatacenter.put(entry.getKey(), counts.allReplicas()); + if (counts.transientReplicas() > 0) { - continue; + 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); + } + /** + * Resolves raw string values, e.g. {@code "3"} or {@code "3/1"}, to per-datacenter counts. + */ + private static Map<String, ReplicaCounts> parse(@NotNull Map<String, String> options) + { + Map<String, ReplicaCounts> parsed = new LinkedHashMap<>(options.size()); + options.forEach((datacenter, value) -> { + if ("class".equals(datacenter)) + { + return; + } try { - this.options.put(entry.getKey(), Integer.parseInt(entry.getValue())); + parsed.put(datacenter, ReplicaCounts.parse(value)); } - catch (NumberFormatException exception) + catch (IllegalArgumentException exception) { - LOGGER.warn("Could not parse replication option: {} = {}", entry.getKey(), entry.getValue()); + throw new IllegalArgumentException(String.format("Could not parse replication option: %s = %s", + datacenter, value), exception); } - } + }); + return parsed; } - public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map<String, Integer> options) + /** + * Builds counts for one datacenter, adding the datacenter name to any validation failure. + * <p> + * {@link ReplicaCounts} is a plain value type and deliberately knows nothing about datacenters, so naming the + * datacenter is the caller's job - see the discussion on CASSANALYTICS-194. + * + * @param datacenter datacenter the counts belong to, used only to describe a failure + * @param allReplicas total replicas + * @param transientReplicas transient (witness) replicas + * @return validated counts + * @throws IllegalArgumentException when the counts are inconsistent, naming the datacenter + */ + private static ReplicaCounts countsFor(String datacenter, int allReplicas, int transientReplicas) { - this.replicationStrategy = replicationStrategy; - this.options = new LinkedHashMap<>(options.size()); - - if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) + try { - throw new RuntimeException(String.format("Could not find replication info in schema map: %s.", options)); + return ReplicaCounts.of(allReplicas, transientReplicas); } - - for (Map.Entry<String, Integer> entry : options.entrySet()) + catch (IllegalArgumentException exception) { - if ("class".equals(entry.getKey())) + throw new IllegalArgumentException( + String.format("%s for datacenter %s", exception.getMessage(), datacenter), exception); + } + } + + /** + * 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<String, ReplicaCounts> merge(@NotNull Map<String, Integer> options, + @NotNull Map<String, Integer> transientOptions) + { + Map<String, ReplicaCounts> merged = new LinkedHashMap<>(options.size()); + options.forEach((datacenter, allReplicas) -> { + if (!"class".equals(datacenter)) { - continue; + merged.put(datacenter, countsFor(datacenter, allReplicas, + transientOptions.getOrDefault(datacenter, 0))); } - this.options.put(entry.getKey(), entry.getValue()); - } + }); + 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; } + /** + * @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() - .mapToInt(Integer::intValue) - .sum(); + return totalReplicationFactor; + } + + /** + * @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 fullReplicationFactor; + } + + /** + * @return the number of transient (witness) replicas across all datacenters, {@code 0} when none are configured + */ + public Integer getTransientReplicationFactor() + { + return transientReplicationFactor; + } + + /** + * @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 it has none + * @throws IllegalArgumentException when {@code datacenter} has no replication factor + */ + public int getTransientReplicas(@NotNull String datacenter) + { + return counts(datacenter).transientReplicas(); + } + + /** + * @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) + { + 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, replication.keySet())); + } + return counts; } + /** + * @return per-datacenter replica counts, unmodifiable + */ + @NotNull + public Map<String, ReplicaCounts> getReplication() + { + return replication; + } + + /** + * @return per-datacenter total replica counts, including transient (witness) replicas. Unmodifiable, and derived + * from {@link #getReplication()}. + */ @NotNull public Map<String, Integer> getOptions() { return options; } + /** + * @return per-datacenter transient (witness) replica counts, unmodifiable. Datacenters without transient + * replicas are absent. + */ + @NotNull + public Map<String, Integer> getTransientOptions() + { + return transientOptions; + } + @NotNull public ReplicationStrategy getReplicationStrategy() { @@ -188,40 +384,228 @@ public class ReplicationFactor implements Serializable ReplicationFactor that = (ReplicationFactor) other; return this.replicationStrategy == that.replicationStrategy - && java.util.Objects.equals(this.options, that.options); + && Objects.equals(this.replication, that.replication); } @Override public int hashCode() { - return Objects.hash(replicationStrategy, options); + return Objects.hash(replicationStrategy, replication); + } + + /** + * {@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 + * @throws InvalidObjectException when deserialized from an incompatible older form + */ + private Object readResolve() throws InvalidObjectException + { + if (replication == null) + { + throw new InvalidObjectException( + "ReplicationFactor was serialized by an incompatible version: per-datacenter replica counts are absent. " + + "Driver and executors must run the same version."); + } + // Rebuild through the canonical constructor so the derived fields, which are transient, are populated + return new ReplicationFactor(replicationStrategy, replication, replication); + } + + /** + * Replica counts for a single datacenter. Cassandra accepts either {@code <replicas>} or + * {@code <replicas>/<transient>}; transient replication predates mutation tracking, and witness replicas + * (CEP-46) reuse the same form. See {@code org.apache.cassandra.locator.ReplicationFactor} in Cassandra. + */ + public static final class ReplicaCounts implements Serializable + { + private static final long serialVersionUID = 2026091500000000001L; + 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 allReplicas total replicas + * @param transientReplicas transient (witness) replicas + * @return validated counts + * @throws IllegalArgumentException when the counts are inconsistent. The message describes the counts + * only; callers that know which datacenter the counts came from are + * responsible for adding that context. + */ + public static ReplicaCounts of(int allReplicas, int transientReplicas) + { + validate(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 + * @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) + { + return of(Integer.parseInt(trimmed), 0); + } + + if (trimmed.indexOf(TRANSIENT_SEPARATOR, separator + 1) >= 0) + { + throw new IllegalArgumentException(String.format( + "Replication factor format is <replicas> or <replicas>/<transient>, found '%s'", value)); + } + + return of(Integer.parseInt(trimmed.substring(0, separator).trim()), + Integer.parseInt(trimmed.substring(separator + 1).trim())); + } + + /** + * 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(int allReplicas, int transientReplicas) + { + if (allReplicas < 0) + { + throw new IllegalArgumentException(String.format( + "Replication factor must be non-negative, found %d", allReplicas)); + } + if (transientReplicas < 0) + { + throw new IllegalArgumentException(String.format( + "Transient replicas must be non-negative, found %d", transientReplicas)); + } + 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", + allReplicas, transientReplicas)); + } + } + + @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<ReplicationFactor> { + /** + * Incremented whenever the Kryo format below changes. Version 1 added the per-datacenter transient + * (witness) replica count, taking each datacenter from one byte to two. + * <p> + * Without the marker an older stream is not merely rejected but silently misread: the reader would take + * the next datacenter's string-length byte as a transient count and desync for the rest of the stream, + * yielding wrong replication factors rather than an error. Kryo is the path Spark uses to ship this to + * executors, so it gets the same protection as the hand-rolled JDK format in + * {@link org.apache.cassandra.spark.data.partitioner.CassandraRing}. + * <p> + * Deliberately outside the {@link ReplicationStrategy#value} range. A pre-versioning stream begins with + * the strategy value, so a version of 0-2 could be mistaken for a valid marker - notably + * {@code SimpleStrategy(1)} - and the stream would misparse anyway. This cannot make an unmarked stream + * safe in general, but it does stop one from being accepted as version 1. + */ + private static final byte SERIALIZATION_FORMAT_VERSION = 0x7F; + @Override public void write(Kryo kryo, Output out, ReplicationFactor replicationFactor) { + out.writeByte(SERIALIZATION_FORMAT_VERSION); out.writeByte(replicationFactor.replicationStrategy.value); - out.writeByte(replicationFactor.options.size()); - for (Map.Entry<String, Integer> entry : replicationFactor.options.entrySet()) + out.writeByte(replicationFactor.replication.size()); + for (Map.Entry<String, ReplicaCounts> entry : replicationFactor.replication.entrySet()) { out.writeString(entry.getKey()); - out.writeByte(entry.getValue()); + out.writeByte(entry.getValue().allReplicas()); + out.writeByte(entry.getValue().transientReplicas()); } } @Override public ReplicationFactor read(Kryo kryo, Input in, Class<ReplicationFactor> type) { + byte formatVersion = in.readByte(); + if (formatVersion != SERIALIZATION_FORMAT_VERSION) + { + throw new IllegalStateException(String.format( + "Unsupported ReplicationFactor Kryo serialization format version %d, expected %d", + formatVersion, SERIALIZATION_FORMAT_VERSION)); + } ReplicationStrategy strategy = ReplicationStrategy.valueOf(in.readByte()); - int numOptions = in.readByte(); - Map<String, Integer> options = new HashMap<>(numOptions); - for (int option = 0; option < numOptions; option++) + int numDatacenters = in.readByte(); + Map<String, ReplicaCounts> replication = new LinkedHashMap<>(numDatacenters); + for (int datacenter = 0; datacenter < numDatacenters; datacenter++) { - options.put(in.readString(), (int) in.readByte()); + String name = in.readString(); + int allReplicas = in.readByte(); + int transientReplicas = in.readByte(); + replication.put(name, countsFor(name, allReplicas, transientReplicas)); } - return new ReplicationFactor(strategy, options); + 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 3d142f8d..5f12e1a8 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 @@ -28,9 +28,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Comparator; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.stream.Collectors; import java.util.Objects; @@ -69,6 +67,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 writes the + * ReplicationFactor as an object rather than destructuring it. + */ + private static final byte SERIALIZATION_FORMAT_VERSION = 1; + private Partitioner partitioner; private String keyspace; private ReplicationFactor replicationFactor; @@ -260,17 +272,18 @@ public class CassandraRing implements Serializable 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(); - ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.valueOf(in.readByte()); - int optionCount = in.readByte(); - Map<String, Integer> options = new HashMap<>(optionCount); - for (int option = 0; option < optionCount; option++) - { - options.put(in.readUTF(), (int) in.readByte()); - } - this.replicationFactor = new ReplicationFactor(strategy, options); + // 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); @@ -284,17 +297,11 @@ public class CassandraRing implements Serializable 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); - out.writeByte(this.replicationFactor.getReplicationStrategy().value); - Map<String, Integer> options = this.replicationFactor.getOptions(); - out.writeByte(options.size()); - for (Map.Entry<String, Integer> option : options.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-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 80d69118..59dc919b 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 final class CqlUtils 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 <replicas>/<transient> form for witness replicas, so delegate parsing to + // 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 new ReplicationFactor(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 ef2eb8ee..1c6cae34 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,15 @@ package org.apache.cassandra.spark.data; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; 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 +118,335 @@ public class ReplicationFactorTests assertThat(replicationFactor1).isEqualTo(replicationFactor2); assertThat(replicationFactor1.hashCode()).isEqualTo(replicationFactor2.hashCode()); } + + // Transient / witness replicas: the <replicas>/<transient> 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 and must not be accepted silently + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/3"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("datacenter1"); + } + + @Test + public void testMalformedReplicationValuesAreRejected() + { + for (String malformed : new String[]{"3/", "/1", "3/1/1", "3/x", "x/1", "3/-1", ""}) + { + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", malformed))) + .as("malformed replication value '%s' should be rejected", malformed) + .isInstanceOf(IllegalArgumentException.class); + } + } + + @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() + { + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "-3"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("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); + } + + @Test + public void testUnparseableValueIsRejected() + { + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "xyz"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("datacenter1"); + } + + @Test + public void testNoDatacenterEntriesIsRejected() + { + assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Could not find replication info"); + } + + @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); + } + + // 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); + } + + @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.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(deserialized.hasTransientReplicas()).isTrue(); + assertThat(deserialized.getOptions()).containsOnlyKeys("datacenter1", "datacenter2"); + assertThat(deserialized.getTransientOptions()).containsOnlyKeys("datacenter1"); + assertThat(deserialized).isEqualTo(original); + } + + @Test + public void testKryoRejectsUnknownSerializationFormatVersion() + { + // Hand-build a stream whose leading version byte is one this reader does not know, then read it through + // the serializer directly. Not via kryo.writeObject/readObject, which wrap the payload in their own + // reference header - corrupting byte 0 of that would test Kryo's framing rather than our version check. + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (Output out = new Output(bytes)) + { + out.writeByte(99); + out.writeByte(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy.value); + out.writeByte(1); + out.writeString("datacenter1"); + out.writeByte(3); + out.writeByte(1); + } + + Kryo kryo = new Kryo(); + ReplicationFactor.Serializer serializer = new ReplicationFactor.Serializer(); + try (Input in = new Input(new ByteArrayInputStream(bytes.toByteArray()))) + { + assertThatThrownBy(() -> serializer.read(kryo, in, ReplicationFactor.class)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Unsupported ReplicationFactor Kryo serialization format version"); + } + } + + @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).isEqualTo(original); + assertThat(deserialized.getTotalReplicationFactor()).isEqualTo(6); + assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5); + assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(deserialized.getTransientReplicas("datacenter2")).isEqualTo(0); + assertThat(deserialized.getOptions()).containsOnlyKeys("datacenter1", "datacenter2"); + assertThat(deserialized.getTransientOptions()).containsOnlyKeys("datacenter1"); + } } 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 c66290f0..a34e499d 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.junit.jupiter.api.Test; 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 class CassandraRingTests 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 58fb5c96..c2cf777a 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,79 @@ public class CqlUtilsTest extends VersionRunner assertThat(systemSchemaRf.getOptions()).isEqualTo(ImmutableMap.of()); } + @Test + public void testExtractReplicationFactorWithWitnessReplicas() + { + String schema = "CREATE KEYSPACE witnessks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3/1', 'datacenter2': '3/1'} 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); + } + + @Test + public void testExtractReplicationFactorMixedWitnessAndFullDatacenters() + { + String schema = "CREATE KEYSPACE mixedks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3/1', 'datacenter2': '3'} AND durable_writes = true;\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 ce51359a..4175ad45 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 class SchemaBuilderTests new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the <replicas>/<transient> 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 d73e1976..92031dd0 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 @@ package org.apache.cassandra.spark.reader; 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 class SchemaBuilderTests .as("build() must preserve enableCdc=false through to the returned CqlTable") .isFalse(); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the <replicas>/<transient> 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 52ea0fd8..723c1533 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 @@ public abstract class AbstractSchemaBuilder result.put("class", "org.apache.cassandra.locator." + replicationFactor.getReplicationStrategy().name()); for (Map.Entry<String, Integer> entry : replicationFactor.getOptions().entrySet()) { + // Deliberately emits only the total, never the <replicas>/<transient> 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-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 f378982c..ba789671 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 class SchemaBuilderTests new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the <replicas>/<transient> 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-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 a8157e0c..93158bf1 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 @@ public abstract class AbstractSchemaBuilder result.put("class", "org.apache.cassandra.locator." + replicationFactor.getReplicationStrategy().name()); for (Map.Entry<String, Integer> entry : replicationFactor.getOptions().entrySet()) { + // Deliberately emits only the total, never the <replicas>/<transient> 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; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
