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;