From 74d94dcd9006057f54fcdadd162e0d6e384935b7 Mon Sep 17 00:00:00 2001 From: mkhara Date: Fri, 28 Aug 2026 15:03:11 -0700 Subject: [PATCH 1/2] 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. Prerequisite for CASSANALYTICS-164. patch by Mansi Khara; reviewed by TBD for CASSANALYTICS-194 --- CHANGES.txt | 1 + .../spark/data/ReplicationFactor.java | 268 +++++++++++++++- .../spark/data/partitioner/CassandraRing.java | 15 +- .../cassandra/spark/utils/CqlUtils.java | 15 +- .../spark/data/ReplicationFactorTests.java | 299 ++++++++++++++++++ .../data/partitioner/CassandraRingTests.java | 44 +++ .../cassandra/spark/utils/CqlUtilsTest.java | 78 +++++ .../spark/reader/SchemaBuilderTests.java | 17 + .../cassandra/spark/reader/SchemaBuilder.java | 4 + 9 files changed, 732 insertions(+), 9 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 19a658dc1..469bbe82b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Parse / replication factor for witness-enabled keyspaces (CASSANALYTICS-194) * Determine whether mutation tracking is enabled for keyspace for bulk writes (CASSANALYTICS-160) * Upgrade sidecar version to 0.4.0 * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java index 0b9b6821d..cb7544faf 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,19 +161,46 @@ 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 = replicationStrategy; this.options = new LinkedHashMap<>(options.size()); + this.transientOptions = new LinkedHashMap<>(transientOptions.size()); if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) { @@ -149,8 +215,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 +244,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 +338,107 @@ 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; + } + return new ReplicationFactor(replicationStrategy, options, Collections.emptyMap()); + } + + /** + * 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 +453,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 +473,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..b8c38766d 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 @@ -270,7 +270,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); @@ -295,6 +301,13 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou out.writeUTF(option.getKey()); out.writeByte(option.getValue()); } + Map transientOptions = this.replicationFactor.getTransientOptions(); + out.writeByte(transientOptions.size()); + for (Map.Entry option : transientOptions.entrySet()) + { + out.writeUTF(option.getKey()); + out.writeByte(option.getValue()); + } out.writeShort(this.instances.size()); for (CassandraInstance instance : this.instances) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java index 8511dbaa1..b76814144 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java @@ -176,9 +176,18 @@ public static ReplicationFactor extractReplicationFactor(@NotNull String schemaS throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), exception); } - String className = map.remove("class"); - ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.getEnum(className); - return new ReplicationFactor(strategy, map.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, v -> Integer.parseInt(v.getValue())))); + // Values may use the / form for witness replicas, so delegate parsing to + // ReplicationFactor. parseStrict reports an unparseable value directly instead of dropping the + // datacenter, which would otherwise surface later as a confusing "DC not found" error. + try + { + return ReplicationFactor.parseStrict(map); + } + catch (IllegalArgumentException exception) + { + throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), + exception); + } } public static String extractTableSchema(@NotNull String schemaStr, @NotNull String keyspace, @NotNull String table) diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java index ef2eb8eeb..ef93969bd 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,296 @@ 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); + } } 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..2bf62476c 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,10 @@ package org.apache.cassandra.spark.data.partitioner; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.math.BigInteger; import java.util.Arrays; import java.util.Collection; @@ -447,4 +451,44 @@ 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. + } } 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-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java index 2c5a2b251..e99c00892 100644 --- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java +++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java @@ -21,6 +21,7 @@ import java.util.HashMap; +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import org.apache.cassandra.bridge.CassandraBridgeImplementation; @@ -114,4 +115,20 @@ public void testSchemaBuilderWithPartiallyInitializedMetadata() new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the / form. The embedded Cassandra runs with + // transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node" for exactly the witness-enabled keyspaces the bulk + // reader needs to read. Dropping it is safe because replica placement comes from CassandraRing. + ReplicationFactor withTransient = new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + assertThat(rfToMap(withTransient)) + .containsEntry("datacenter1", "3") + .containsEntry("datacenter2", "3"); + } } diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java index 5d0493a1f..736f89a04 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java @@ -586,6 +586,10 @@ static Map rfToMap(ReplicationFactor replicationFactor) result.put("class", "org.apache.cassandra.locator." + replicationFactor.getReplicationStrategy().name()); for (Map.Entry entry : replicationFactor.getOptions().entrySet()) { + // Deliberately emits only the total, never the / form. The embedded Cassandra runs + // with transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node". Safe to drop because replica placement comes from + // CassandraRing, not from these KeyspaceParams. result.put(entry.getKey(), Integer.toString(entry.getValue())); } return result; From da8100b3c1012a129a5508537b0aaeb8b097df3b Mon Sep 17 00:00:00 2001 From: mkhara Date: Fri, 28 Aug 2026 15:03:11 -0700 Subject: [PATCH 2/2] 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. Prerequisite for CASSANALYTICS-164. patch by Mansi Khara; reviewed by TBD for CASSANALYTICS-194 --- CHANGES.txt | 1 + .../spark/data/ReplicationFactor.java | 268 +++++++++++++++- .../spark/data/partitioner/CassandraRing.java | 15 +- .../cassandra/spark/utils/CqlUtils.java | 15 +- .../spark/data/ReplicationFactorTests.java | 299 ++++++++++++++++++ .../data/partitioner/CassandraRingTests.java | 44 +++ .../cassandra/spark/utils/CqlUtilsTest.java | 78 +++++ .../spark/reader/SchemaBuilderTests.java | 16 + .../spark/reader/SchemaBuilderTests.java | 17 + .../cassandra/spark/reader/SchemaBuilder.java | 4 + 10 files changed, 748 insertions(+), 9 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 19a658dc1..469bbe82b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Parse / replication factor for witness-enabled keyspaces (CASSANALYTICS-194) * Determine whether mutation tracking is enabled for keyspace for bulk writes (CASSANALYTICS-160) * Upgrade sidecar version to 0.4.0 * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java index 0b9b6821d..cb7544faf 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,19 +161,46 @@ 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 = replicationStrategy; this.options = new LinkedHashMap<>(options.size()); + this.transientOptions = new LinkedHashMap<>(transientOptions.size()); if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) { @@ -149,8 +215,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 +244,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 +338,107 @@ 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; + } + return new ReplicationFactor(replicationStrategy, options, Collections.emptyMap()); + } + + /** + * 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 +453,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 +473,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..b8c38766d 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 @@ -270,7 +270,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); @@ -295,6 +301,13 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou out.writeUTF(option.getKey()); out.writeByte(option.getValue()); } + Map transientOptions = this.replicationFactor.getTransientOptions(); + out.writeByte(transientOptions.size()); + for (Map.Entry option : transientOptions.entrySet()) + { + out.writeUTF(option.getKey()); + out.writeByte(option.getValue()); + } out.writeShort(this.instances.size()); for (CassandraInstance instance : this.instances) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java index 8511dbaa1..b76814144 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java @@ -176,9 +176,18 @@ public static ReplicationFactor extractReplicationFactor(@NotNull String schemaS throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), exception); } - String className = map.remove("class"); - ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.getEnum(className); - return new ReplicationFactor(strategy, map.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, v -> Integer.parseInt(v.getValue())))); + // Values may use the / form for witness replicas, so delegate parsing to + // ReplicationFactor. parseStrict reports an unparseable value directly instead of dropping the + // datacenter, which would otherwise surface later as a confusing "DC not found" error. + try + { + return ReplicationFactor.parseStrict(map); + } + catch (IllegalArgumentException exception) + { + throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), + exception); + } } public static String extractTableSchema(@NotNull String schemaStr, @NotNull String keyspace, @NotNull String table) diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java index ef2eb8eeb..ef93969bd 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,296 @@ 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); + } } 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..2bf62476c 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,10 @@ package org.apache.cassandra.spark.data.partitioner; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.math.BigInteger; import java.util.Arrays; import java.util.Collection; @@ -447,4 +451,44 @@ 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. + } } diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java index 58fb5c960..3a237a695 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java @@ -142,6 +142,84 @@ public void testExtractReplicationFactor(CassandraBridge bridge) assertThat(systemSchemaRf.getOptions()).isEqualTo(ImmutableMap.of()); } + @Test + public void testExtractReplicationFactorWithWitnessReplicas() + { + // Witness-enabled keyspace as created by Cassandra's WitnessAlwaysReadsFullReplicaTest on the + // cep-45-mutation-tracking branch: the / form plus replication_type = 'tracked' + String schema = "CREATE KEYSPACE witnessks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3/1', 'datacenter2': '3/1'} AND replication_type = 'tracked' " + + "AND durable_writes = true;\n"; + + ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "witnessks"); + assertThat(rf).isNotNull(); + assertThat(rf.getReplicationStrategy()).isEqualTo(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy); + // total keeps its original meaning: all replicas, witnesses included + assertThat(rf.getOptions()).isEqualTo(ImmutableMap.of("datacenter1", 3, "datacenter2", 3)); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(6); + assertThat(rf.getFullReplicationFactor()).isEqualTo(4); + assertThat(rf.getTransientReplicationFactor()).isEqualTo(2); + assertThat(rf.hasTransientReplicas()).isTrue(); + assertThat(rf.getFullReplicas("datacenter1")).isEqualTo(2); + assertThat(rf.getTransientReplicas("datacenter1")).isEqualTo(1); + + assertThat(CqlUtils.isTracked(CqlUtils.extractReplicationType(schema, "witnessks"))).isTrue(); + } + + @Test + public void testExtractReplicationFactorMixedWitnessAndFullDatacenters() + { + String schema = "CREATE KEYSPACE mixedks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3/1', 'datacenter2': '3'} AND replication_type = 'tracked';\n"; + + ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "mixedks"); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(6); + assertThat(rf.getFullReplicationFactor()).isEqualTo(5); + assertThat(rf.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(rf.getTransientReplicas("datacenter2")).isEqualTo(0); + } + + @Test + public void testExtractReplicationFactorUntrackedKeyspaceUnaffected() + { + String schema = "CREATE KEYSPACE plainks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '3'} AND durable_writes = true;\n"; + + ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema, "plainks"); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(3); + assertThat(rf.getFullReplicationFactor()).isEqualTo(3); + assertThat(rf.hasTransientReplicas()).isFalse(); + assertThat(rf.getTransientOptions()).isEmpty(); + } + + @Test + public void testExtractReplicationFactorFailsLoudlyOnUnparseableValue() + { + // An unparseable value must be reported here rather than silently dropping the datacenter, which + // would surface later as a confusing "DC not found in replication factor" error + for (String malformed : new String[]{ "xyz", "3/", "3/1/1", "3/x", "3/3" }) + { + String schema = "CREATE KEYSPACE badks WITH REPLICATION = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1': '" + malformed + "'} AND durable_writes = true;\n"; + assertThatThrownBy(() -> CqlUtils.extractReplicationFactor(schema, "badks")) + .as("malformed replication value '%s' should raise", malformed) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Unable to parse replication factor for keyspace: badks"); + } + } + + @Test + public void testExtractReplicationFactorFailsLoudlyOnMissingDatacenterEntries() + { + // A NetworkTopologyStrategy keyspace with no datacenter entries is not usable. It must fail here + // rather than yielding an empty replication factor that fails later with "DC not found" + String schema = "CREATE KEYSPACE emptyks WITH REPLICATION = {'class': 'NetworkTopologyStrategy'} " + + "AND durable_writes = true;\n"; + assertThatThrownBy(() -> CqlUtils.extractReplicationFactor(schema, "emptyks")) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Unable to parse replication factor for keyspace: emptyks"); + } + @ParameterizedTest @MethodSource("org.apache.cassandra.spark.data.VersionRunner#bridges") public void testEscapedColumnNames(CassandraBridge bridge) diff --git a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java index ce51359a6..4175ad457 100644 --- a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java +++ b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java @@ -115,4 +115,20 @@ public void testSchemaBuilderWithPartiallyInitializedMetadata() new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the / form. The embedded Cassandra runs with + // transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node" for exactly the witness-enabled keyspaces the bulk + // reader needs to read. Dropping it is safe because replica placement comes from CassandraRing. + ReplicationFactor withTransient = new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + assertThat(rfToMap(withTransient)) + .containsEntry("datacenter1", "3") + .containsEntry("datacenter2", "3"); + } } diff --git a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java index 2c5a2b251..e99c00892 100644 --- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java +++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java @@ -21,6 +21,7 @@ import java.util.HashMap; +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import org.apache.cassandra.bridge.CassandraBridgeImplementation; @@ -114,4 +115,20 @@ public void testSchemaBuilderWithPartiallyInitializedMetadata() new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testRfToMapOmitsTransientReplicas() + { + // rfToMap must emit only the total, never the / form. The embedded Cassandra runs with + // transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node" for exactly the witness-enabled keyspaces the bulk + // reader needs to read. Dropping it is safe because replica placement comes from CassandraRing. + ReplicationFactor withTransient = new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + assertThat(rfToMap(withTransient)) + .containsEntry("datacenter1", "3") + .containsEntry("datacenter2", "3"); + } } diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java index 5d0493a1f..736f89a04 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java @@ -586,6 +586,10 @@ static Map rfToMap(ReplicationFactor replicationFactor) result.put("class", "org.apache.cassandra.locator." + replicationFactor.getReplicationStrategy().name()); for (Map.Entry entry : replicationFactor.getOptions().entrySet()) { + // Deliberately emits only the total, never the / form. The embedded Cassandra runs + // with transient_replication_enabled=false, so a transient value makes Keyspace.openWithoutSSTables throw + // "Transient replication is not enabled on this node". Safe to drop because replica placement comes from + // CassandraRing, not from these KeyspaceParams. result.put(entry.getKey(), Integer.toString(entry.getValue())); } return result;