This is an automated email from the ASF dual-hosted git repository.
sarankk pushed a commit to branch mutation-tracking-support
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/mutation-tracking-support by
this push:
new 941b58fc CASSANALYTICS-194: Parse <replicas>/<transient> replication
factor for witness-enabled keyspaces (#238)
941b58fc is described below
commit 941b58fc5b02803995c8a75944b726e7bc92c0ed
Author: Mansi Khara <[email protected]>
AuthorDate: Thu Oct 1 19:46:06 2026 -0400
CASSANALYTICS-194: Parse <replicas>/<transient> replication factor for
witness-enabled keyspaces (#238)
Patch by Mansi Khara; Reviewed by Saranya Krishnakumar, Jyothsna Konisa,
Bernardo Botella for CASSANALYTICS-194
---
CHANGES.txt | 1 +
.../cassandra/spark/data/ReplicationFactor.java | 462 +++++++++++++++++++--
.../spark/data/partitioner/CassandraRing.java | 43 +-
.../org/apache/cassandra/spark/utils/CqlUtils.java | 15 +-
.../spark/data/ReplicationFactorTests.java | 338 +++++++++++++++
.../spark/data/partitioner/CassandraRingTests.java | 104 +++++
.../apache/cassandra/spark/utils/CqlUtilsTest.java | 73 ++++
.../cassandra/spark/reader/SchemaBuilderTests.java | 16 +
.../cassandra/spark/reader/SchemaBuilderTests.java | 17 +
.../spark/reader/AbstractSchemaBuilder.java | 4 +
.../cassandra/spark/reader/SchemaBuilderTests.java | 16 +
.../spark/reader/AbstractSchemaBuilder.java | 4 +
12 files changed, 1033 insertions(+), 60 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 95fe46d6..dbb7415e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Parse <replicas>/<transient> replication factor for witness-enabled
keyspaces (CASSANALYTICS-194)
* Determine whether mutation tracking is enabled for keyspace for bulk writes
(CASSANALYTICS-160)
* Upgrade sidecar version to 0.4.0
* Opening one keyspace instance rebuilds every keyspace instance in the
Cassandra 6.0 bridge (CASSANALYTICS-202)
diff --git
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java
index 0b9b6821..9abc88ad 100644
---
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java
+++
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java
@@ -19,16 +19,15 @@
package org.apache.cassandra.spark.data;
+import java.io.InvalidObjectException;
import java.io.Serializable;
-import java.util.HashMap;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import com.google.common.collect.ImmutableMap;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
@@ -50,12 +49,16 @@ import org.jetbrains.annotations.NotNull;
* "replication_factor" : 1
* }
* }
+ * <p>
+ * Replica counts may also use the {@code <replicas>/<transient>} form, e.g.
{@code "DC1" : "3/1"}, meaning three
+ * replicas of which one is transient. Witness replicas (CEP-46) reuse this
form, so {@code "3/1"} describes two full
+ * replicas and one witness. {@link #getTotalReplicationFactor()} continues to
report all three; use
+ * {@link #getFullReplicationFactor()} for the count that holds the full data
set.
*/
public class ReplicationFactor implements Serializable
{
public static final Serializer SERIALIZER = new Serializer();
private static final long serialVersionUID = -2017022813595983257L;
- private static final Logger LOGGER =
LoggerFactory.getLogger(ReplicationFactor.class);
public enum ReplicationStrategy
{
@@ -106,64 +109,257 @@ public class ReplicationFactor implements Serializable
@NotNull
private final ReplicationStrategy replicationStrategy;
+ /**
+ * Per-datacenter replica counts. Holding the total and the transient
count together, rather than in two
+ * parallel maps, means the two cannot drift apart.
+ */
@NotNull
- private final Map<String, Integer> options;
+ private final Map<String, ReplicaCounts> replication;
+ // Derived from replication, computed once because this class is
immutable. Marked transient so they are absent
+ // from the serialized form: readResolve rebuilds through the canonical
constructor, and the Kryo serializer
+ // already goes through it, so neither path can leave these out of step
with replication.
+ private final transient int totalReplicationFactor;
+ private final transient int fullReplicationFactor;
+ private final transient int transientReplicationFactor;
+ private final transient Map<String, Integer> options;
+ private final transient Map<String, Integer> transientOptions;
+
+ /**
+ * Parses a raw replication map. A value that cannot be parsed, or that
Cassandra itself would reject, raises
+ * {@link IllegalArgumentException} naming the offending datacenter: a
partial replication factor is not usable
+ * and reporting it here avoids a misleading failure later, such as "DC
not found in replication factor".
+ *
+ * @param options the raw replication map, including the {@code class}
entry
+ * @throws IllegalArgumentException when any replication value cannot be
parsed
+ */
public ReplicationFactor(@NotNull Map<String, String> options)
{
- this.replicationStrategy =
ReplicationFactor.ReplicationStrategy.getEnum(options.get("class"));
- this.options = new LinkedHashMap<>(options.size());
- for (Map.Entry<String, String> entry : options.entrySet())
+ this(ReplicationStrategy.getEnum(options.get("class")),
parse(options), options);
+ }
+
+ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
@NotNull Map<String, Integer> options)
+ {
+ this(replicationStrategy, options, Collections.emptyMap());
+ }
+
+ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
+ @NotNull Map<String, Integer> options,
+ @NotNull Map<String, Integer> transientOptions)
+ {
+ this(replicationStrategy, merge(options, transientOptions), options);
+ }
+
+ /**
+ * Canonical constructor. Every other constructor resolves its input to
per-datacenter counts and delegates here.
+ *
+ * @param replicationStrategy the replication strategy
+ * @param replication per-datacenter replica counts, already
validated individually
+ * @param raw the caller's original input, used only in
the error message
+ */
+ private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
+ @NotNull Map<String, ReplicaCounts> replication,
+ @NotNull Object raw)
+ {
+ // A strategy other than LocalStrategy with no datacenter entries is
not usable
+ if (replicationStrategy != ReplicationStrategy.LocalStrategy &&
replication.isEmpty())
+ {
+ throw new IllegalArgumentException("Could not find replication
info in schema map: " + raw);
+ }
+ this.replicationStrategy = replicationStrategy;
+ this.replication = Collections.unmodifiableMap(new
LinkedHashMap<>(replication));
+
+ int total = 0;
+ int transientCount = 0;
+ Map<String, Integer> allByDatacenter = new
LinkedHashMap<>(this.replication.size());
+ Map<String, Integer> transientByDatacenter = new LinkedHashMap<>();
+ for (Map.Entry<String, ReplicaCounts> entry :
this.replication.entrySet())
{
- if ("class".equals(entry.getKey()))
+ ReplicaCounts counts = entry.getValue();
+ total += counts.allReplicas();
+ transientCount += counts.transientReplicas();
+ allByDatacenter.put(entry.getKey(), counts.allReplicas());
+ if (counts.transientReplicas() > 0)
{
- continue;
+ transientByDatacenter.put(entry.getKey(),
counts.transientReplicas());
}
+ }
+ this.totalReplicationFactor = total;
+ this.transientReplicationFactor = transientCount;
+ this.fullReplicationFactor = total - transientCount;
+ this.options = Collections.unmodifiableMap(allByDatacenter);
+ this.transientOptions =
Collections.unmodifiableMap(transientByDatacenter);
+ }
+ /**
+ * Resolves raw string values, e.g. {@code "3"} or {@code "3/1"}, to
per-datacenter counts.
+ */
+ private static Map<String, ReplicaCounts> parse(@NotNull Map<String,
String> options)
+ {
+ Map<String, ReplicaCounts> parsed = new
LinkedHashMap<>(options.size());
+ options.forEach((datacenter, value) -> {
+ if ("class".equals(datacenter))
+ {
+ return;
+ }
try
{
- this.options.put(entry.getKey(),
Integer.parseInt(entry.getValue()));
+ parsed.put(datacenter, ReplicaCounts.parse(value));
}
- catch (NumberFormatException exception)
+ catch (IllegalArgumentException exception)
{
- LOGGER.warn("Could not parse replication option: {} = {}",
entry.getKey(), entry.getValue());
+ throw new IllegalArgumentException(String.format("Could not
parse replication option: %s = %s",
+ datacenter,
value), exception);
}
- }
+ });
+ return parsed;
}
- public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
@NotNull Map<String, Integer> options)
+ /**
+ * Builds counts for one datacenter, adding the datacenter name to any
validation failure.
+ * <p>
+ * {@link ReplicaCounts} is a plain value type and deliberately knows
nothing about datacenters, so naming the
+ * datacenter is the caller's job - see the discussion on
CASSANALYTICS-194.
+ *
+ * @param datacenter datacenter the counts belong to, used only to
describe a failure
+ * @param allReplicas total replicas
+ * @param transientReplicas transient (witness) replicas
+ * @return validated counts
+ * @throws IllegalArgumentException when the counts are inconsistent,
naming the datacenter
+ */
+ private static ReplicaCounts countsFor(String datacenter, int allReplicas,
int transientReplicas)
{
- this.replicationStrategy = replicationStrategy;
- this.options = new LinkedHashMap<>(options.size());
-
- if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) &&
options.isEmpty())
+ try
{
- throw new RuntimeException(String.format("Could not find
replication info in schema map: %s.", options));
+ return ReplicaCounts.of(allReplicas, transientReplicas);
}
-
- for (Map.Entry<String, Integer> entry : options.entrySet())
+ catch (IllegalArgumentException exception)
{
- if ("class".equals(entry.getKey()))
+ throw new IllegalArgumentException(
+ String.format("%s for datacenter %s", exception.getMessage(),
datacenter), exception);
+ }
+ }
+
+ /**
+ * Resolves separate total and transient maps to per-datacenter counts,
rejecting a transient entry for a
+ * datacenter that has no replication factor.
+ */
+ private static Map<String, ReplicaCounts> merge(@NotNull Map<String,
Integer> options,
+ @NotNull Map<String,
Integer> transientOptions)
+ {
+ Map<String, ReplicaCounts> merged = new
LinkedHashMap<>(options.size());
+ options.forEach((datacenter, allReplicas) -> {
+ if (!"class".equals(datacenter))
{
- continue;
+ merged.put(datacenter, countsFor(datacenter, allReplicas,
+
transientOptions.getOrDefault(datacenter, 0)));
}
- this.options.put(entry.getKey(), entry.getValue());
- }
+ });
+ transientOptions.forEach((datacenter, transientReplicas) -> {
+ if (!"class".equals(datacenter) && transientReplicas != null &&
transientReplicas != 0
+ && !merged.containsKey(datacenter))
+ {
+ throw new IllegalArgumentException(String.format(
+ "Transient replicas specified for %s but it has no replication
factor", datacenter));
+ }
+ });
+ return merged;
}
+ /**
+ * @return the total number of replicas across all datacenters, including
transient (witness) replicas.
+ * Semantics are unchanged from before transient replica support
was added.
+ */
public Integer getTotalReplicationFactor()
{
- return options.values().stream()
- .mapToInt(Integer::intValue)
- .sum();
+ return totalReplicationFactor;
+ }
+
+ /**
+ * @return the number of replicas across all datacenters that hold the
full data set, i.e. the total
+ * replication factor minus transient (witness) replicas
+ */
+ public Integer getFullReplicationFactor()
+ {
+ return fullReplicationFactor;
+ }
+
+ /**
+ * @return the number of transient (witness) replicas across all
datacenters, {@code 0} when none are configured
+ */
+ public Integer getTransientReplicationFactor()
+ {
+ return transientReplicationFactor;
+ }
+
+ /**
+ * @return {@code true} if any datacenter is configured with transient
(witness) replicas
+ */
+ public boolean hasTransientReplicas()
+ {
+ return !transientOptions.isEmpty();
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of transient (witness) replicas in {@code
datacenter}, {@code 0} when it has none
+ * @throws IllegalArgumentException when {@code datacenter} has no
replication factor
+ */
+ public int getTransientReplicas(@NotNull String datacenter)
+ {
+ return counts(datacenter).transientReplicas();
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of replicas in {@code datacenter} holding the full
data set
+ * @throws IllegalArgumentException when {@code datacenter} has no
replication factor
+ */
+ public int getFullReplicas(@NotNull String datacenter)
+ {
+ return counts(datacenter).fullReplicas();
+ }
+
+ private ReplicaCounts counts(@NotNull String datacenter)
+ {
+ ReplicaCounts counts = replication.get(datacenter);
+ if (counts == null)
+ {
+ throw new IllegalArgumentException(String.format("Datacenter %s
not found in replication factor %s",
+ datacenter,
replication.keySet()));
+ }
+ return counts;
}
+ /**
+ * @return per-datacenter replica counts, unmodifiable
+ */
+ @NotNull
+ public Map<String, ReplicaCounts> getReplication()
+ {
+ return replication;
+ }
+
+ /**
+ * @return per-datacenter total replica counts, including transient
(witness) replicas. Unmodifiable, and derived
+ * from {@link #getReplication()}.
+ */
@NotNull
public Map<String, Integer> getOptions()
{
return options;
}
+ /**
+ * @return per-datacenter transient (witness) replica counts,
unmodifiable. Datacenters without transient
+ * replicas are absent.
+ */
+ @NotNull
+ public Map<String, Integer> getTransientOptions()
+ {
+ return transientOptions;
+ }
+
@NotNull
public ReplicationStrategy getReplicationStrategy()
{
@@ -188,40 +384,228 @@ public class ReplicationFactor implements Serializable
ReplicationFactor that = (ReplicationFactor) other;
return this.replicationStrategy == that.replicationStrategy
- && java.util.Objects.equals(this.options, that.options);
+ && Objects.equals(this.replication, that.replication);
}
@Override
public int hashCode()
{
- return Objects.hash(replicationStrategy, options);
+ return Objects.hash(replicationStrategy, replication);
+ }
+
+ /**
+ * {@link #serialVersionUID} is pinned, so an instance serialized before
the per-datacenter counts were combined
+ * deserializes with a {@code null} {@code replication}. Only reachable
under driver/executor version skew, which
+ * is not a supported configuration, so this reports the mismatch rather
than silently losing the counts.
+ *
+ * @return this instance
+ * @throws InvalidObjectException when deserialized from an incompatible
older form
+ */
+ private Object readResolve() throws InvalidObjectException
+ {
+ if (replication == null)
+ {
+ throw new InvalidObjectException(
+ "ReplicationFactor was serialized by an incompatible version:
per-datacenter replica counts are absent. "
+ + "Driver and executors must run the same version.");
+ }
+ // Rebuild through the canonical constructor so the derived fields,
which are transient, are populated
+ return new ReplicationFactor(replicationStrategy, replication,
replication);
+ }
+
+ /**
+ * Replica counts for a single datacenter. Cassandra accepts either {@code
<replicas>} or
+ * {@code <replicas>/<transient>}; transient replication predates mutation
tracking, and witness replicas
+ * (CEP-46) reuse the same form. See {@code
org.apache.cassandra.locator.ReplicationFactor} in Cassandra.
+ */
+ public static final class ReplicaCounts implements Serializable
+ {
+ private static final long serialVersionUID = 2026091500000000001L;
+ private static final String TRANSIENT_SEPARATOR = "/";
+
+ private final int allReplicas;
+ private final int transientReplicas;
+
+ private ReplicaCounts(int allReplicas, int transientReplicas)
+ {
+ this.allReplicas = allReplicas;
+ this.transientReplicas = transientReplicas;
+ }
+
+ /**
+ * @param allReplicas total replicas
+ * @param transientReplicas transient (witness) replicas
+ * @return validated counts
+ * @throws IllegalArgumentException when the counts are inconsistent.
The message describes the counts
+ * only; callers that know which
datacenter the counts came from are
+ * responsible for adding that context.
+ */
+ public static ReplicaCounts of(int allReplicas, int transientReplicas)
+ {
+ validate(allReplicas, transientReplicas);
+ return new ReplicaCounts(allReplicas, transientReplicas);
+ }
+
+ /**
+ * @return total replicas, including transient (witness) replicas
+ */
+ public int allReplicas()
+ {
+ return allReplicas;
+ }
+
+ /**
+ * @return replicas holding the full data set
+ */
+ public int fullReplicas()
+ {
+ return allReplicas - transientReplicas;
+ }
+
+ /**
+ * @return transient (witness) replicas, {@code 0} when none are
configured
+ */
+ public int transientReplicas()
+ {
+ return transientReplicas;
+ }
+
+ /**
+ * @param value the raw replication value, e.g. {@code "3"} or {@code
"3/1"}
+ * @return the parsed replica counts
+ * @throws NumberFormatException when either component is not an
integer
+ * @throws IllegalArgumentException when the value is malformed or the
counts are inconsistent
+ */
+ static ReplicaCounts parse(@NotNull String value)
+ {
+ String trimmed = value.trim();
+ int separator = trimmed.indexOf(TRANSIENT_SEPARATOR);
+ if (separator < 0)
+ {
+ return of(Integer.parseInt(trimmed), 0);
+ }
+
+ if (trimmed.indexOf(TRANSIENT_SEPARATOR, separator + 1) >= 0)
+ {
+ throw new IllegalArgumentException(String.format(
+ "Replication factor format is <replicas> or
<replicas>/<transient>, found '%s'", value));
+ }
+
+ return of(Integer.parseInt(trimmed.substring(0, separator).trim()),
+ Integer.parseInt(trimmed.substring(separator +
1).trim()));
+ }
+
+ /**
+ * Mirrors the constraints Cassandra enforces in {@code
ReplicationFactor.validate}: transient replicas must
+ * be non-negative and strictly fewer than the total, so at least one
full replica always exists.
+ *
+ * @param datacenter datacenter name for the error message, may
be {@code null}
+ * @param allReplicas total replicas
+ * @param transientReplicas transient (witness) replicas
+ */
+ static void validate(int allReplicas, int transientReplicas)
+ {
+ if (allReplicas < 0)
+ {
+ throw new IllegalArgumentException(String.format(
+ "Replication factor must be non-negative, found %d",
allReplicas));
+ }
+ if (transientReplicas < 0)
+ {
+ throw new IllegalArgumentException(String.format(
+ "Transient replicas must be non-negative, found %d",
transientReplicas));
+ }
+ if (transientReplicas > 0 && transientReplicas >= allReplicas)
+ {
+ throw new IllegalArgumentException(String.format(
+ "Transient replicas must be zero, or less than the total
replication factor. For %d/%d",
+ allReplicas, transientReplicas));
+ }
+ }
+
+ @Override
+ public boolean equals(Object other)
+ {
+ if (this == other)
+ {
+ return true;
+ }
+ if (other == null || getClass() != other.getClass())
+ {
+ return false;
+ }
+ ReplicaCounts that = (ReplicaCounts) other;
+ return allReplicas == that.allReplicas && transientReplicas ==
that.transientReplicas;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(allReplicas, transientReplicas);
+ }
+
+ @Override
+ public String toString()
+ {
+ return transientReplicas > 0 ? allReplicas + TRANSIENT_SEPARATOR +
transientReplicas
+ : String.valueOf(allReplicas);
+ }
}
public static class Serializer extends
com.esotericsoftware.kryo.Serializer<ReplicationFactor>
{
+ /**
+ * Incremented whenever the Kryo format below changes. Version 1 added
the per-datacenter transient
+ * (witness) replica count, taking each datacenter from one byte to
two.
+ * <p>
+ * Without the marker an older stream is not merely rejected but
silently misread: the reader would take
+ * the next datacenter's string-length byte as a transient count and
desync for the rest of the stream,
+ * yielding wrong replication factors rather than an error. Kryo is
the path Spark uses to ship this to
+ * executors, so it gets the same protection as the hand-rolled JDK
format in
+ * {@link org.apache.cassandra.spark.data.partitioner.CassandraRing}.
+ * <p>
+ * Deliberately outside the {@link ReplicationStrategy#value} range. A
pre-versioning stream begins with
+ * the strategy value, so a version of 0-2 could be mistaken for a
valid marker - notably
+ * {@code SimpleStrategy(1)} - and the stream would misparse anyway.
This cannot make an unmarked stream
+ * safe in general, but it does stop one from being accepted as
version 1.
+ */
+ private static final byte SERIALIZATION_FORMAT_VERSION = 0x7F;
+
@Override
public void write(Kryo kryo, Output out, ReplicationFactor
replicationFactor)
{
+ out.writeByte(SERIALIZATION_FORMAT_VERSION);
out.writeByte(replicationFactor.replicationStrategy.value);
- out.writeByte(replicationFactor.options.size());
- for (Map.Entry<String, Integer> entry :
replicationFactor.options.entrySet())
+ out.writeByte(replicationFactor.replication.size());
+ for (Map.Entry<String, ReplicaCounts> entry :
replicationFactor.replication.entrySet())
{
out.writeString(entry.getKey());
- out.writeByte(entry.getValue());
+ out.writeByte(entry.getValue().allReplicas());
+ out.writeByte(entry.getValue().transientReplicas());
}
}
@Override
public ReplicationFactor read(Kryo kryo, Input in,
Class<ReplicationFactor> type)
{
+ byte formatVersion = in.readByte();
+ if (formatVersion != SERIALIZATION_FORMAT_VERSION)
+ {
+ throw new IllegalStateException(String.format(
+ "Unsupported ReplicationFactor Kryo serialization format
version %d, expected %d",
+ formatVersion, SERIALIZATION_FORMAT_VERSION));
+ }
ReplicationStrategy strategy =
ReplicationStrategy.valueOf(in.readByte());
- int numOptions = in.readByte();
- Map<String, Integer> options = new HashMap<>(numOptions);
- for (int option = 0; option < numOptions; option++)
+ int numDatacenters = in.readByte();
+ Map<String, ReplicaCounts> replication = new
LinkedHashMap<>(numDatacenters);
+ for (int datacenter = 0; datacenter < numDatacenters; datacenter++)
{
- options.put(in.readString(), (int) in.readByte());
+ String name = in.readString();
+ int allReplicas = in.readByte();
+ int transientReplicas = in.readByte();
+ replication.put(name, countsFor(name, allReplicas,
transientReplicas));
}
- return new ReplicationFactor(strategy, options);
+ return new ReplicationFactor(strategy, replication, replication);
}
}
}
diff --git
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
index 3d142f8d..5f12e1a8 100644
---
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
+++
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
@@ -28,9 +28,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.stream.Collectors;
import java.util.Objects;
@@ -69,6 +67,20 @@ public class CassandraRing implements Serializable
private static final Logger LOGGER =
LoggerFactory.getLogger(CassandraRing.class);
public static final Serializer SERIALIZER = new Serializer();
+ /**
+ * Pinned so that the JDK serialization format change made when transient
(witness) replica counts were added is
+ * detected. Without it the UID is computed from the class signature,
which did not change - only the
+ * {@link #readObject}/{@link #writeObject} bodies did - so an older
stream would be silently misread rather than
+ * rejected.
+ */
+ private static final long serialVersionUID = 2026082800000000001L;
+
+ /**
+ * Incremented whenever the hand-rolled JDK serialization format below
changes. Version 1 writes the
+ * ReplicationFactor as an object rather than destructuring it.
+ */
+ private static final byte SERIALIZATION_FORMAT_VERSION = 1;
+
private Partitioner partitioner;
private String keyspace;
private ReplicationFactor replicationFactor;
@@ -260,17 +272,18 @@ public class CassandraRing implements Serializable
private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException
{
LOGGER.debug("Falling back to JDK deserialization");
+ byte formatVersion = in.readByte();
+ if (formatVersion != SERIALIZATION_FORMAT_VERSION)
+ {
+ throw new IOException(String.format("Unsupported CassandraRing
serialization format version %d, expected %d",
+ formatVersion,
SERIALIZATION_FORMAT_VERSION));
+ }
this.partitioner = in.readByte() == 0 ? Partitioner.RandomPartitioner
: Partitioner.Murmur3Partitioner;
this.keyspace = in.readUTF();
- ReplicationFactor.ReplicationStrategy strategy =
ReplicationFactor.ReplicationStrategy.valueOf(in.readByte());
- int optionCount = in.readByte();
- Map<String, Integer> options = new HashMap<>(optionCount);
- for (int option = 0; option < optionCount; option++)
- {
- options.put(in.readUTF(), (int) in.readByte());
- }
- this.replicationFactor = new ReplicationFactor(strategy, options);
+ // ReplicationFactor is Serializable, so it is written whole rather
than destructured. That keeps this
+ // method independent of how ReplicationFactor represents its
per-datacenter counts.
+ this.replicationFactor = (ReplicationFactor) in.readObject();
int numInstances = in.readShort();
this.instances = new ArrayList<>(numInstances);
@@ -284,17 +297,11 @@ public class CassandraRing implements Serializable
private void writeObject(ObjectOutputStream out) throws IOException,
ClassNotFoundException
{
LOGGER.debug("Falling back to JDK serialization");
+ out.writeByte(SERIALIZATION_FORMAT_VERSION);
out.writeByte(this.partitioner == Partitioner.RandomPartitioner ? 0 :
1);
out.writeUTF(this.keyspace);
- out.writeByte(this.replicationFactor.getReplicationStrategy().value);
- Map<String, Integer> options = this.replicationFactor.getOptions();
- out.writeByte(options.size());
- for (Map.Entry<String, Integer> option : options.entrySet())
- {
- out.writeUTF(option.getKey());
- out.writeByte(option.getValue());
- }
+ out.writeObject(this.replicationFactor);
out.writeShort(this.instances.size());
for (CassandraInstance instance : this.instances)
diff --git
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java
index 80d69118..59dc919b 100644
---
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java
+++
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java
@@ -176,9 +176,18 @@ public final class CqlUtils
throw new RuntimeException(String.format("Unable to parse
replication factor for keyspace: %s", keyspace), exception);
}
- String className = map.remove("class");
- ReplicationFactor.ReplicationStrategy strategy =
ReplicationFactor.ReplicationStrategy.getEnum(className);
- return new ReplicationFactor(strategy,
map.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, v ->
Integer.parseInt(v.getValue()))));
+ // Values may use the <replicas>/<transient> form for witness
replicas, so delegate parsing to
+ // ReplicationFactor, which reports an unparseable value rather than
dropping the datacenter. Dropping it
+ // would surface later as a confusing "DC not found in replication
factor" error.
+ try
+ {
+ return new ReplicationFactor(map);
+ }
+ catch (IllegalArgumentException exception)
+ {
+ throw new RuntimeException(String.format("Unable to parse
replication factor for keyspace: %s", keyspace),
+ exception);
+ }
}
public static String extractTableSchema(@NotNull String schemaStr,
@NotNull String keyspace, @NotNull String table)
diff --git
a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java
b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java
index ef2eb8ee..1c6cae34 100644
---
a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java
+++
b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java
@@ -19,8 +19,15 @@
package org.apache.cassandra.spark.data;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
import java.util.ArrayList;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
import com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Test;
@@ -111,4 +118,335 @@ public class ReplicationFactorTests
assertThat(replicationFactor1).isEqualTo(replicationFactor2);
assertThat(replicationFactor1.hashCode()).isEqualTo(replicationFactor2.hashCode());
}
+
+ // Transient / witness replicas: the <replicas>/<transient> form, reused
by witness replicas under
+ // mutation tracking (CEP-45/CEP-46)
+
+ @Test
+ public void testNoTransientReplicasByDefault()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3"));
+ assertThat(replicationFactor.hasTransientReplicas()).isFalse();
+ assertThat(replicationFactor.getTransientOptions()).isEmpty();
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(3);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(0);
+
assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(3);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(0);
+ }
+
+ @Test
+ public void testTransientReplicasSingleDatacenter()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThat(replicationFactor.hasTransientReplicas()).isTrue();
+ // total keeps its original meaning: all replicas, witnesses included
+
assertThat(replicationFactor.getOptions().get("datacenter1")).isEqualTo(Integer.valueOf(3));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(1);
+
assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(2);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1);
+ }
+
+ @Test
+ public void testTransientReplicasMultipleDatacenters()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3/1"));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(4);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testMixedTransientAndFullDatacenters()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(5);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1);
+
assertThat(replicationFactor.getTransientReplicas("datacenter2")).isEqualTo(0);
+
assertThat(replicationFactor.getFullReplicas("datacenter2")).isEqualTo(3);
+ // datacenter2 has no transient replicas, so it must be absent rather
than mapped to zero
+
assertThat(replicationFactor.getTransientOptions()).containsOnlyKeys("datacenter1");
+ }
+
+ @Test
+ public void testTransientReplicasSimpleStrategy()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "SimpleStrategy",
+ "replication_factor", "3/1"));
+ assertThat(replicationFactor.getReplicationStrategy())
+ .isEqualTo(ReplicationFactor.ReplicationStrategy.SimpleStrategy);
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testTransientReplicasWithWhitespace()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", " 3 / 1 "));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testTransientEqualToTotalIsRejected()
+ {
+ // Cassandra requires at least one full replica, so 3/3 is invalid and
must not be accepted silently
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/3")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("datacenter1");
+ }
+
+ @Test
+ public void testMalformedReplicationValuesAreRejected()
+ {
+ for (String malformed : new String[]{"3/", "/1", "3/1/1", "3/x",
"x/1", "3/-1", ""})
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", malformed)))
+ .as("malformed replication value '%s' should be rejected",
malformed)
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+
+ @Test
+ public void testGetFullReplicasUnknownDatacenter()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThatThrownBy(() -> replicationFactor.getFullReplicas("nosuchdc"))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testTransientOptionsForUnknownDatacenterRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(
+ ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy,
+ ImmutableMap.of("datacenter1", 3),
+ ImmutableMap.of("datacenter2", 1)))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testEqualityConsidersTransientReplicas()
+ {
+ ReplicationFactor full = new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3"));
+ ReplicationFactor withTransient = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThat(full).isNotEqualTo(withTransient);
+ assertThat(full.hashCode()).isNotEqualTo(withTransient.hashCode());
+ }
+
+ @Test
+ public void testNegativeReplicationFactorIsRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "-3")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("datacenter1");
+ }
+
+ @Test
+ public void testZeroReplicationFactorIsAllowed()
+ {
+ // RF 0 is legitimate for NetworkTopologyStrategy: the keyspace is
simply not replicated to that datacenter
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3",
+ "datacenter2", "0"));
+
assertThat(replicationFactor.getOptions().get("datacenter2")).isEqualTo(Integer.valueOf(0));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ }
+
+ @Test
+ public void testUnparseableValueIsRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "xyz")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("datacenter1");
+ }
+
+ @Test
+ public void testNoDatacenterEntriesIsRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Could not find replication info");
+ }
+
+ @Test
+ public void testLocalStrategyWithNoEntriesIsAllowed()
+ {
+ // LocalStrategy legitimately has no datacenter entries, e.g. the
system_schema keyspace
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "org.apache.cassandra.locator.LocalStrategy"));
+ assertThat(replicationFactor.getReplicationStrategy())
+ .isEqualTo(ReplicationFactor.ReplicationStrategy.LocalStrategy);
+ assertThat(replicationFactor.getOptions()).isEmpty();
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(0);
+ }
+
+ // The per-datacenter counts are held together rather than in two parallel
maps, so the total and the
+ // transient count cannot drift apart
+
+ @Test
+ public void testReplicationExposesCombinedCounts()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+
+
assertThat(replicationFactor.getReplication()).containsOnlyKeys("datacenter1",
"datacenter2");
+ ReplicationFactor.ReplicaCounts dc1 =
replicationFactor.getReplication().get("datacenter1");
+ assertThat(dc1.allReplicas()).isEqualTo(3);
+ assertThat(dc1.fullReplicas()).isEqualTo(2);
+ assertThat(dc1.transientReplicas()).isEqualTo(1);
+
+ ReplicationFactor.ReplicaCounts dc2 =
replicationFactor.getReplication().get("datacenter2");
+ assertThat(dc2.allReplicas()).isEqualTo(3);
+ assertThat(dc2.fullReplicas()).isEqualTo(3);
+ assertThat(dc2.transientReplicas()).isEqualTo(0);
+ }
+
+ @Test
+ public void testDerivedOptionMapsAreUnmodifiable()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThatThrownBy(() ->
replicationFactor.getOptions().put("datacenter2", 3))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() ->
replicationFactor.getTransientOptions().put("datacenter2", 1))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() ->
replicationFactor.getReplication().remove("datacenter1"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ @Test
+ public void testGetTransientReplicasUnknownDatacenter()
+ {
+ // Consistent with getFullReplicas: an unknown datacenter is a
programming error, not zero transients
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThatThrownBy(() ->
replicationFactor.getTransientReplicas("nosuchdc"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("nosuchdc");
+ }
+
+ @Test
+ public void testPrecomputedValuesSurviveJdkSerialization() throws Exception
+ {
+ // The aggregate counts and derived maps are precomputed and
transient, so they are absent from the
+ // serialized form and must be rebuilt on the way back in
+ ReplicationFactor original = new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream out = new ObjectOutputStream(bytes))
+ {
+ out.writeObject(original);
+ }
+ ReplicationFactor deserialized;
+ try (ObjectInputStream in = new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray())))
+ {
+ deserialized = (ReplicationFactor) in.readObject();
+ }
+
+ assertThat(deserialized.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5);
+ assertThat(deserialized.getTransientReplicationFactor()).isEqualTo(1);
+
assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1);
+ assertThat(deserialized.hasTransientReplicas()).isTrue();
+ assertThat(deserialized.getOptions()).containsOnlyKeys("datacenter1",
"datacenter2");
+
assertThat(deserialized.getTransientOptions()).containsOnlyKeys("datacenter1");
+ assertThat(deserialized).isEqualTo(original);
+ }
+
+ @Test
+ public void testKryoRejectsUnknownSerializationFormatVersion()
+ {
+ // Hand-build a stream whose leading version byte is one this reader
does not know, then read it through
+ // the serializer directly. Not via kryo.writeObject/readObject, which
wrap the payload in their own
+ // reference header - corrupting byte 0 of that would test Kryo's
framing rather than our version check.
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (Output out = new Output(bytes))
+ {
+ out.writeByte(99);
+
out.writeByte(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy.value);
+ out.writeByte(1);
+ out.writeString("datacenter1");
+ out.writeByte(3);
+ out.writeByte(1);
+ }
+
+ Kryo kryo = new Kryo();
+ ReplicationFactor.Serializer serializer = new
ReplicationFactor.Serializer();
+ try (Input in = new Input(new
ByteArrayInputStream(bytes.toByteArray())))
+ {
+ assertThatThrownBy(() -> serializer.read(kryo, in,
ReplicationFactor.class))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Unsupported ReplicationFactor Kryo
serialization format version");
+ }
+ }
+
+ @Test
+ public void testPrecomputedValuesSurviveKryoSerialization()
+ {
+ ReplicationFactor original = new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+
+ Kryo kryo = new Kryo();
+ kryo.register(ReplicationFactor.class, new
ReplicationFactor.Serializer());
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (Output out = new Output(bytes))
+ {
+ kryo.writeObject(out, original);
+ }
+ ReplicationFactor deserialized;
+ try (Input in = new Input(new
ByteArrayInputStream(bytes.toByteArray())))
+ {
+ deserialized = kryo.readObject(in, ReplicationFactor.class);
+ }
+
+ assertThat(deserialized).isEqualTo(original);
+ assertThat(deserialized.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5);
+
assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1);
+
assertThat(deserialized.getTransientReplicas("datacenter2")).isEqualTo(0);
+ assertThat(deserialized.getOptions()).containsOnlyKeys("datacenter1",
"datacenter2");
+
assertThat(deserialized.getTransientOptions()).containsOnlyKeys("datacenter1");
+ }
}
diff --git
a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java
b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java
index c66290f0..a34e499d 100644
---
a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java
+++
b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java
@@ -19,6 +19,11 @@
package org.apache.cassandra.spark.data.partitioner;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
import java.math.BigInteger;
import java.util.Arrays;
import java.util.Collection;
@@ -34,6 +39,7 @@ import org.junit.jupiter.api.Test;
import org.apache.cassandra.spark.data.ReplicationFactor;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
@SuppressWarnings("UnstableApiUsage")
public class CassandraRingTests
@@ -447,4 +453,102 @@ public class CassandraRingTests
Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken()));
}
+
+ @Test
+ public void testJdkSerializationPreservesTransientReplicas() throws
Exception
+ {
+ // CassandraRing hand-rolls readObject/writeObject and rebuilds
ReplicationFactor from strategy plus
+ // options, so a transient (witness) count could silently vanish on
the way to a Spark executor
+ CassandraRing ring = new CassandraRing(
+ Partitioner.Murmur3Partitioner,
+ "test",
+ new ReplicationFactor(ImmutableMap.of("class",
"org.apache.cassandra.locator.NetworkTopologyStrategy",
+ "DC1", "3/1",
+ "DC2", "3")),
+ Arrays.asList(new CassandraInstance("0", "local0-i1", "DC1"),
+ new CassandraInstance("100", "local0-i2", "DC1"),
+ new CassandraInstance("200", "local0-i3", "DC1"),
+ new CassandraInstance("1", "local1-i1", "DC2"),
+ new CassandraInstance("101", "local1-i2", "DC2"),
+ new CassandraInstance("201", "local1-i3", "DC2")));
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream out = new ObjectOutputStream(bytes))
+ {
+ out.writeObject(ring);
+ }
+ CassandraRing deserialized;
+ try (ObjectInputStream in = new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray())))
+ {
+ deserialized = (CassandraRing) in.readObject();
+ }
+
+ ReplicationFactor rf = deserialized.replicationFactor();
+ assertThat(rf.getTransientReplicas("DC1")).isEqualTo(1);
+ assertThat(rf.getTransientReplicas("DC2")).isEqualTo(0);
+ assertThat(rf.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(rf.getFullReplicationFactor()).isEqualTo(5);
+ assertThat(rf).isEqualTo(ring.replicationFactor());
+ // Deliberately not asserting deserialized.equals(ring):
CassandraRing#equals compares the derived
+ // replicas and tokenRangeMap fields, and does not hold across a JDK
round trip even without transient
+ // replicas. Pre-existing behaviour, unrelated to replica types.
+ }
+
+ @Test
+ public void testJdkDeserializationRejectsUnknownFormatVersion() throws
Exception
+ {
+ // The hand-rolled format gained a transient-replica section, but the
class signature did not change, so the
+ // computed serialVersionUID would still match an older stream. A
leading format-version byte makes a stale
+ // stream fail loudly instead of being misread.
+ CassandraRing ring = new CassandraRing(
+ Partitioner.Murmur3Partitioner,
+ "test",
+ new ReplicationFactor(ImmutableMap.of("class",
"org.apache.cassandra.locator.NetworkTopologyStrategy",
+ "DC1", "3")),
+ Arrays.asList(new CassandraInstance("0", "local0-i1", "DC1"),
+ new CassandraInstance("100", "local0-i2", "DC1"),
+ new CassandraInstance("200", "local0-i3", "DC1")));
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream out = new ObjectOutputStream(bytes))
+ {
+ out.writeObject(ring);
+ }
+
+ // Locate the format-version byte deterministically: the format is
+ // [version][partitioner] then writeUTF(keyspace), so it sits two
bytes before the "test" UTF header
+ byte[] raw = bytes.toByteArray();
+ byte[] keyspaceUtf = new byte[]{0, 4, 't', 'e', 's', 't'};
+ int keyspaceIndex = indexOf(raw, keyspaceUtf);
+ assertThat(keyspaceIndex).as("keyspace UTF header should be
locatable").isGreaterThan(1);
+ int versionIndex = keyspaceIndex - 2;
+ assertThat(raw[versionIndex]).as("format version
byte").isEqualTo((byte) 1);
+
+ // Simulate a stream written before the transient-replica section
existed
+ raw[versionIndex] = 0;
+
+ try (ObjectInputStream in = new ObjectInputStream(new
ByteArrayInputStream(raw)))
+ {
+ assertThatThrownBy(in::readObject)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported CassandraRing serialization
format version");
+ }
+ }
+
+ private static int indexOf(byte[] haystack, byte[] needle)
+ {
+ outer:
+ for (int i = 0; i <= haystack.length - needle.length; i++)
+ {
+ for (int j = 0; j < needle.length; j++)
+ {
+ if (haystack[i + j] != needle[j])
+ {
+ continue outer;
+ }
+ }
+ return i;
+ }
+ return -1;
+ }
}
diff --git
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java
index 58fb5c96..c2cf777a 100644
---
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java
+++
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java
@@ -142,6 +142,79 @@ public class CqlUtilsTest extends VersionRunner
assertThat(systemSchemaRf.getOptions()).isEqualTo(ImmutableMap.of());
}
+ @Test
+ public void testExtractReplicationFactorWithWitnessReplicas()
+ {
+ String schema = "CREATE KEYSPACE witnessks WITH REPLICATION =
{'class': 'NetworkTopologyStrategy', "
+ + "'datacenter1': '3/1', 'datacenter2': '3/1'} AND
durable_writes = true;\n";
+
+ ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema,
"witnessks");
+ assertThat(rf).isNotNull();
+
assertThat(rf.getReplicationStrategy()).isEqualTo(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy);
+ // total keeps its original meaning: all replicas, witnesses included
+ assertThat(rf.getOptions()).isEqualTo(ImmutableMap.of("datacenter1",
3, "datacenter2", 3));
+ assertThat(rf.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(rf.getFullReplicationFactor()).isEqualTo(4);
+ assertThat(rf.getTransientReplicationFactor()).isEqualTo(2);
+ assertThat(rf.hasTransientReplicas()).isTrue();
+ assertThat(rf.getFullReplicas("datacenter1")).isEqualTo(2);
+ assertThat(rf.getTransientReplicas("datacenter1")).isEqualTo(1);
+ }
+
+ @Test
+ public void testExtractReplicationFactorMixedWitnessAndFullDatacenters()
+ {
+ String schema = "CREATE KEYSPACE mixedks WITH REPLICATION = {'class':
'NetworkTopologyStrategy', "
+ + "'datacenter1': '3/1', 'datacenter2': '3'} AND
durable_writes = true;\n";
+
+ ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema,
"mixedks");
+ assertThat(rf.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(rf.getFullReplicationFactor()).isEqualTo(5);
+ assertThat(rf.getTransientReplicas("datacenter1")).isEqualTo(1);
+ assertThat(rf.getTransientReplicas("datacenter2")).isEqualTo(0);
+ }
+
+ @Test
+ public void testExtractReplicationFactorUntrackedKeyspaceUnaffected()
+ {
+ String schema = "CREATE KEYSPACE plainks WITH REPLICATION = {'class':
'NetworkTopologyStrategy', "
+ + "'datacenter1': '3'} AND durable_writes = true;\n";
+
+ ReplicationFactor rf = CqlUtils.extractReplicationFactor(schema,
"plainks");
+ assertThat(rf.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(rf.getFullReplicationFactor()).isEqualTo(3);
+ assertThat(rf.hasTransientReplicas()).isFalse();
+ assertThat(rf.getTransientOptions()).isEmpty();
+ }
+
+ @Test
+ public void testExtractReplicationFactorFailsLoudlyOnUnparseableValue()
+ {
+ // An unparseable value must be reported here rather than silently
dropping the datacenter, which
+ // would surface later as a confusing "DC not found in replication
factor" error
+ for (String malformed : new String[]{"xyz", "3/", "3/1/1", "3/x",
"3/3"})
+ {
+ String schema = "CREATE KEYSPACE badks WITH REPLICATION =
{'class': 'NetworkTopologyStrategy', "
+ + "'datacenter1': '" + malformed + "'} AND
durable_writes = true;\n";
+ assertThatThrownBy(() -> CqlUtils.extractReplicationFactor(schema,
"badks"))
+ .as("malformed replication value '%s' should raise", malformed)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to parse replication factor for
keyspace: badks");
+ }
+ }
+
+ @Test
+ public void
testExtractReplicationFactorFailsLoudlyOnMissingDatacenterEntries()
+ {
+ // A NetworkTopologyStrategy keyspace with no datacenter entries is
not usable. It must fail here
+ // rather than yielding an empty replication factor that fails later
with "DC not found"
+ String schema = "CREATE KEYSPACE emptyks WITH REPLICATION = {'class':
'NetworkTopologyStrategy'} "
+ + "AND durable_writes = true;\n";
+ assertThatThrownBy(() -> CqlUtils.extractReplicationFactor(schema,
"emptyks"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to parse replication factor for
keyspace: emptyks");
+ }
+
@ParameterizedTest
@MethodSource("org.apache.cassandra.spark.data.VersionRunner#bridges")
public void testEscapedColumnNames(CassandraBridge bridge)
diff --git
a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
index ce51359a..4175ad45 100644
---
a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
+++
b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
@@ -115,4 +115,20 @@ public class SchemaBuilderTests
new SchemaBuilder(createTableStatement, keyspaceName,
replicationFactor);
}
+
+ @Test
+ public void testRfToMapOmitsTransientReplicas()
+ {
+ // rfToMap must emit only the total, never the <replicas>/<transient>
form. The embedded Cassandra runs with
+ // transient_replication_enabled=false, so a transient value makes
Keyspace.openWithoutSSTables throw
+ // "Transient replication is not enabled on this node" for exactly the
witness-enabled keyspaces the bulk
+ // reader needs to read. Dropping it is safe because replica placement
comes from CassandraRing.
+ ReplicationFactor withTransient = new
ReplicationFactor(ImmutableMap.of(
+ "class", "org.apache.cassandra.locator.NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+ assertThat(rfToMap(withTransient))
+ .containsEntry("datacenter1", "3")
+ .containsEntry("datacenter2", "3");
+ }
}
diff --git
a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
index d73e1976..92031dd0 100644
---
a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
+++
b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
@@ -22,6 +22,7 @@ package org.apache.cassandra.spark.reader;
import java.util.Collections;
import java.util.HashMap;
+import com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Test;
import org.apache.cassandra.bridge.CassandraBridgeImplementation;
@@ -164,4 +165,20 @@ public class SchemaBuilderTests
.as("build() must preserve enableCdc=false through to the
returned CqlTable")
.isFalse();
}
+
+ @Test
+ public void testRfToMapOmitsTransientReplicas()
+ {
+ // rfToMap must emit only the total, never the <replicas>/<transient>
form. The embedded Cassandra runs with
+ // transient_replication_enabled=false, so a transient value makes
Keyspace.openWithoutSSTables throw
+ // "Transient replication is not enabled on this node" for exactly the
witness-enabled keyspaces the bulk
+ // reader needs to read. Dropping it is safe because replica placement
comes from CassandraRing.
+ ReplicationFactor withTransient = new
ReplicationFactor(ImmutableMap.of(
+ "class", "org.apache.cassandra.locator.NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+ assertThat(rfToMap(withTransient))
+ .containsEntry("datacenter1", "3")
+ .containsEntry("datacenter2", "3");
+ }
}
diff --git
a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
index 52ea0fd8..723c1533 100644
---
a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
+++
b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
@@ -589,6 +589,10 @@ public abstract class AbstractSchemaBuilder
result.put("class", "org.apache.cassandra.locator." +
replicationFactor.getReplicationStrategy().name());
for (Map.Entry<String, Integer> entry :
replicationFactor.getOptions().entrySet())
{
+ // Deliberately emits only the total, never the
<replicas>/<transient> form. The embedded Cassandra runs
+ // with transient_replication_enabled=false, so a transient value
makes Keyspace.openWithoutSSTables throw
+ // "Transient replication is not enabled on this node". Safe to
drop because replica placement comes from
+ // CassandraRing, not from these KeyspaceParams.
result.put(entry.getKey(), Integer.toString(entry.getValue()));
}
return result;
diff --git
a/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
b/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
index f378982c..ba789671 100644
---
a/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
+++
b/cassandra-six-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java
@@ -138,4 +138,20 @@ public class SchemaBuilderTests
new SchemaBuilder(createTableStatement, keyspaceName,
replicationFactor);
}
+
+ @Test
+ public void testRfToMapOmitsTransientReplicas()
+ {
+ // rfToMap must emit only the total, never the <replicas>/<transient>
form. The embedded Cassandra runs with
+ // transient_replication_enabled=false, so a transient value makes
Keyspace.openWithoutSSTables throw
+ // "Transient replication is not enabled on this node" for exactly the
witness-enabled keyspaces the bulk
+ // reader needs to read. Dropping it is safe because replica placement
comes from CassandraRing.
+ ReplicationFactor withTransient = new
ReplicationFactor(ImmutableMap.of(
+ "class", "org.apache.cassandra.locator.NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+ assertThat(rfToMap(withTransient))
+ .containsEntry("datacenter1", "3")
+ .containsEntry("datacenter2", "3");
+ }
}
diff --git
a/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
b/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
index a8157e0c..93158bf1 100644
---
a/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
+++
b/cassandra-six-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java
@@ -589,6 +589,10 @@ public abstract class AbstractSchemaBuilder
result.put("class", "org.apache.cassandra.locator." +
replicationFactor.getReplicationStrategy().name());
for (Map.Entry<String, Integer> entry :
replicationFactor.getOptions().entrySet())
{
+ // Deliberately emits only the total, never the
<replicas>/<transient> form. The embedded Cassandra runs
+ // with transient_replication_enabled=false, so a transient value
makes Keyspace.openWithoutSSTables throw
+ // "Transient replication is not enabled on this node". Safe to
drop because replica placement comes from
+ // CassandraRing, not from these KeyspaceParams.
result.put(entry.getKey(), Integer.toString(entry.getValue()));
}
return result;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
