diff --git a/CHANGES.txt b/CHANGES.txt index 974b9ceaa..dda6945a6 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,8 @@ 0.5.0 ----- + * Source token ranges from Cassandra for mutation tracked keyspaces (CASSANALYTICS-197) + * 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) * Regenerate bloom filters for CQLSSTableWriter (CASSANALYTICS-167) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java index 0b9b6821d..ef40bcad8 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java @@ -20,6 +20,7 @@ package org.apache.cassandra.spark.data; import java.io.Serializable; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; @@ -50,6 +51,11 @@ * "replication_factor" : 1 * } * } + *

+ * Replica counts may also use the {@code /} form, e.g. {@code "DC1" : "3/1"}, meaning three + * replicas of which one is transient. Witness replicas under mutation tracking (CEP-45/CEP-46) reuse this form, so + * {@code "3/1"} describes two full replicas and one witness. {@link #getTotalReplicationFactor()} continues to + * report all three; use {@link #getFullReplicationFactor()} for the count that holds the full data set. */ public class ReplicationFactor implements Serializable { @@ -108,11 +114,44 @@ public static ReplicationFactor simpleStrategy(int rf) private final ReplicationStrategy replicationStrategy; @NotNull private final Map options; + /** + * Per-datacenter count of transient (witness) replicas, parsed from the {@code /} form. + * A datacenter absent from this map has no transient replicas. Always empty for untracked keyspaces using the + * plain {@code } form, which keeps behaviour identical for those keyspaces. + */ + @NotNull + private final Map transientOptions; + /** + * Lenient parse: a replication value that cannot be parsed is logged and its datacenter omitted. Retained for + * callers that tolerate a partial replication factor. + * + * @param options the raw replication map, including the {@code class} entry + */ public ReplicationFactor(@NotNull Map options) + { + this(options, false); + } + + /** + * Strict parse: a replication value that cannot be parsed raises {@link IllegalArgumentException} naming the + * offending datacenter, rather than silently omitting it. Prefer this when a partial replication factor would + * produce a misleading failure later. + * + * @param options the raw replication map, including the {@code class} entry + * @return the parsed replication factor + * @throws IllegalArgumentException when any replication value cannot be parsed + */ + public static ReplicationFactor parseStrict(@NotNull Map options) + { + return new ReplicationFactor(options, true); + } + + private ReplicationFactor(@NotNull Map options, boolean strict) { this.replicationStrategy = ReplicationFactor.ReplicationStrategy.getEnum(options.get("class")); this.options = new LinkedHashMap<>(options.size()); + this.transientOptions = new LinkedHashMap<>(); for (Map.Entry entry : options.entrySet()) { if ("class".equals(entry.getKey())) @@ -122,21 +161,56 @@ public ReplicationFactor(@NotNull Map options) try { - this.options.put(entry.getKey(), Integer.parseInt(entry.getValue())); + ReplicaCounts counts = ReplicaCounts.parse(entry.getValue()); + this.options.put(entry.getKey(), counts.allReplicas); + if (counts.transientReplicas > 0) + { + this.transientOptions.put(entry.getKey(), counts.transientReplicas); + } } - catch (NumberFormatException exception) + catch (IllegalArgumentException exception) { + if (strict) + { + throw new IllegalArgumentException(String.format("Could not parse replication option: %s = %s", + entry.getKey(), entry.getValue()), exception); + } LOGGER.warn("Could not parse replication option: {} = {}", entry.getKey(), entry.getValue()); } } + + // Mirrors the guard on the (strategy, options) constructor. A strategy other than LocalStrategy with no + // datacenter entries is not usable, and reporting it here keeps the failure at parse time rather than + // surfacing later as a misleading "DC not found in replication factor". Strict-only, so the lenient + // constructor's behaviour is unchanged for callers that tolerate a partial replication factor. + if (strict && replicationStrategy != ReplicationStrategy.LocalStrategy && this.options.isEmpty()) + { + throw new IllegalArgumentException("Could not find replication info in schema map: " + options); + } } public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotNull Map options) + { + this(replicationStrategy, options, Collections.emptyMap()); + } + + public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, + @NotNull Map options, + @NotNull Map transientOptions) + { + this(replicationStrategy, options, transientOptions, true); + } + + private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, + @NotNull Map options, + @NotNull Map transientOptions, + boolean validate) { this.replicationStrategy = replicationStrategy; this.options = new LinkedHashMap<>(options.size()); + this.transientOptions = new LinkedHashMap<>(transientOptions.size()); - if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) + if (validate && !replicationStrategy.equals(ReplicationStrategy.LocalStrategy) && options.isEmpty()) { throw new RuntimeException(String.format("Could not find replication info in schema map: %s.", options)); } @@ -149,8 +223,28 @@ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy, @NotN } this.options.put(entry.getKey(), entry.getValue()); } + + for (Map.Entry entry : transientOptions.entrySet()) + { + if (entry.getValue() == null || entry.getValue() == 0) + { + continue; + } + Integer allReplicas = this.options.get(entry.getKey()); + if (allReplicas == null) + { + throw new IllegalArgumentException(String.format( + "Transient replicas specified for %s but it has no replication factor", entry.getKey())); + } + ReplicaCounts.validate(entry.getKey(), allReplicas, entry.getValue()); + this.transientOptions.put(entry.getKey(), entry.getValue()); + } } + /** + * @return the total number of replicas across all datacenters, including transient (witness) replicas. + * Semantics are unchanged from before transient replica support was added. + */ public Integer getTotalReplicationFactor() { return options.values().stream() @@ -158,12 +252,76 @@ public Integer getTotalReplicationFactor() .sum(); } + /** + * @return the number of replicas across all datacenters that hold the full data set, i.e. the total + * replication factor minus transient (witness) replicas + */ + public Integer getFullReplicationFactor() + { + return getTotalReplicationFactor() - getTransientReplicationFactor(); + } + + /** + * @return the number of transient (witness) replicas across all datacenters, {@code 0} when none are configured + */ + public Integer getTransientReplicationFactor() + { + return transientOptions.values().stream() + .mapToInt(Integer::intValue) + .sum(); + } + + /** + * @return {@code true} if any datacenter is configured with transient (witness) replicas + */ + public boolean hasTransientReplicas() + { + return !transientOptions.isEmpty(); + } + + /** + * @param datacenter the datacenter to look up + * @return the number of transient (witness) replicas in {@code datacenter}, {@code 0} when none are configured + */ + public int getTransientReplicas(@NotNull String datacenter) + { + return transientOptions.getOrDefault(datacenter, 0); + } + + /** + * @param datacenter the datacenter to look up + * @return the number of replicas in {@code datacenter} holding the full data set + * @throws IllegalArgumentException when {@code datacenter} has no replication factor + */ + public int getFullReplicas(@NotNull String datacenter) + { + Integer allReplicas = options.get(datacenter); + if (allReplicas == null) + { + throw new IllegalArgumentException(String.format("Datacenter %s not found in replication factor %s", + datacenter, options.keySet())); + } + return allReplicas - getTransientReplicas(datacenter); + } + + /** + * @return per-datacenter total replica counts, including transient (witness) replicas + */ @NotNull public Map getOptions() { return options; } + /** + * @return per-datacenter transient (witness) replica counts. Datacenters without transient replicas are absent. + */ + @NotNull + public Map getTransientOptions() + { + return transientOptions; + } + @NotNull public ReplicationStrategy getReplicationStrategy() { @@ -188,13 +346,110 @@ public boolean equals(Object other) ReplicationFactor that = (ReplicationFactor) other; return this.replicationStrategy == that.replicationStrategy - && java.util.Objects.equals(this.options, that.options); + && java.util.Objects.equals(this.options, that.options) + && java.util.Objects.equals(this.transientOptions, that.transientOptions); } @Override public int hashCode() { - return Objects.hash(replicationStrategy, options); + return Objects.hash(replicationStrategy, options, transientOptions); + } + + /** + * {@link #serialVersionUID} is pinned, so an instance serialized before transient replica support was added + * deserializes with a {@code null} {@code transientOptions}. Normalise it to an empty map so the accessors do not + * NPE. Only reachable under driver/executor version skew, which is not a supported configuration. + * + * @return this instance, or a normalised copy when deserialized from the older form + */ + private Object readResolve() + { + if (transientOptions != null) + { + return this; + } + // Bypasses validation deliberately: the older lenient constructor permitted a non-LocalStrategy + // replication factor with no datacenter entries, so re-validating here would turn a legacy instance + // into a deserialization failure + return new ReplicationFactor(replicationStrategy, options, Collections.emptyMap(), false); + } + + /** + * Parsed form of a single datacenter's replication value. Cassandra accepts either {@code } or + * {@code /}, the latter reused by witness replicas under mutation tracking. See + * {@code org.apache.cassandra.locator.ReplicationFactor} in Cassandra. + */ + private static final class ReplicaCounts + { + private static final String TRANSIENT_SEPARATOR = "/"; + + private final int allReplicas; + private final int transientReplicas; + + private ReplicaCounts(int allReplicas, int transientReplicas) + { + this.allReplicas = allReplicas; + this.transientReplicas = transientReplicas; + } + + /** + * @param value the raw replication value, e.g. {@code "3"} or {@code "3/1"} + * @return the parsed replica counts + * @throws NumberFormatException when either component is not an integer + * @throws IllegalArgumentException when the value is malformed or the counts are inconsistent + */ + static ReplicaCounts parse(@NotNull String value) + { + String trimmed = value.trim(); + int separator = trimmed.indexOf(TRANSIENT_SEPARATOR); + if (separator < 0) + { + int allReplicas = Integer.parseInt(trimmed); + validate(null, allReplicas, 0); + return new ReplicaCounts(allReplicas, 0); + } + + if (trimmed.indexOf(TRANSIENT_SEPARATOR, separator + 1) >= 0) + { + throw new IllegalArgumentException(String.format( + "Replication factor format is or /, found '%s'", value)); + } + + int allReplicas = Integer.parseInt(trimmed.substring(0, separator).trim()); + int transientReplicas = Integer.parseInt(trimmed.substring(separator + 1).trim()); + validate(null, allReplicas, transientReplicas); + return new ReplicaCounts(allReplicas, transientReplicas); + } + + /** + * Mirrors the constraints Cassandra enforces in {@code ReplicationFactor.validate}: transient replicas must + * be non-negative and strictly fewer than the total, so at least one full replica always exists. + * + * @param datacenter datacenter name for the error message, may be {@code null} + * @param allReplicas total replicas + * @param transientReplicas transient (witness) replicas + */ + static void validate(String datacenter, int allReplicas, int transientReplicas) + { + String where = datacenter == null ? "" : String.format(" for datacenter %s", datacenter); + if (allReplicas < 0) + { + throw new IllegalArgumentException(String.format( + "Replication factor must be non-negative, found %d%s", allReplicas, where)); + } + if (transientReplicas < 0) + { + throw new IllegalArgumentException(String.format( + "Transient replicas must be non-negative, found %d%s", transientReplicas, where)); + } + if (transientReplicas > 0 && transientReplicas >= allReplicas) + { + throw new IllegalArgumentException(String.format( + "Transient replicas must be zero, or less than the total replication factor. For %d/%d%s", + allReplicas, transientReplicas, where)); + } + } } public static class Serializer extends com.esotericsoftware.kryo.Serializer @@ -209,6 +464,14 @@ public void write(Kryo kryo, Output out, ReplicationFactor replicationFactor) out.writeString(entry.getKey()); out.writeByte(entry.getValue()); } + // Transient (witness) replica counts, written after the totals so the common + // no-transient-replicas case costs a single zero byte + out.writeByte(replicationFactor.transientOptions.size()); + for (Map.Entry entry : replicationFactor.transientOptions.entrySet()) + { + out.writeString(entry.getKey()); + out.writeByte(entry.getValue()); + } } @Override @@ -221,7 +484,13 @@ public ReplicationFactor read(Kryo kryo, Input in, Class type { options.put(in.readString(), (int) in.readByte()); } - return new ReplicationFactor(strategy, options); + int numTransientOptions = in.readByte(); + Map transientOptions = new HashMap<>(numTransientOptions); + for (int option = 0; option < numTransientOptions; option++) + { + transientOptions.put(in.readString(), (int) in.readByte()); + } + return new ReplicationFactor(strategy, options, transientOptions); } } } diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java index 3d142f8df..d5e461922 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 @@ -58,7 +58,10 @@ * As Cassandra token ranges are dependent on Replication strategy, ring makes sense for a specific keyspace only. * It is made to be immutable for the sake of simplicity. *

- * Token ranges are calculated assuming Cassandra racks are not being used, but controlled by assigning tokens properly. + * Token ranges are either supplied by Cassandra, or calculated locally. The local calculation assumes Cassandra racks + * are not being used, but controlled by assigning tokens properly. Callers that cannot rely on that assumption, such + * as the bulk reader for a mutation tracked keyspace, supply the ranges instead - see + * {@link #CassandraRing(Partitioner, String, ReplicationFactor, Collection, Map)}. *

* {@link #equals(Object)} and {@link #hashCode()} don't take {@link #replicas} and {@link #tokenRangeMap} * into consideration as they are just derived fields. @@ -69,10 +72,34 @@ public class CassandraRing implements Serializable private static final Logger LOGGER = LoggerFactory.getLogger(CassandraRing.class); public static final Serializer SERIALIZER = new Serializer(); + /** + * Pinned so that the JDK serialization format change made when transient (witness) replica counts were added is + * detected. Without it the UID is computed from the class signature, which did not change - only the + * {@link #readObject}/{@link #writeObject} bodies did - so an older stream would be silently misread rather than + * rejected. + */ + private static final long serialVersionUID = 2026082800000000001L; + + /** + * Incremented whenever the hand-rolled JDK serialization format below changes. Version 1 added the + * per-datacenter transient (witness) replica counts after the replication options. Version 2 added the + * optional Cassandra-reported token ranges after the instances. + */ + private static final byte SERIALIZATION_FORMAT_VERSION = 2; + private Partitioner partitioner; private String keyspace; private ReplicationFactor replicationFactor; private List instances; + /** + * Token ranges supplied by Cassandra rather than derived locally from tokens and replication factor. + * {@code null} means derive, which is the behaviour for every caller that predates witness replica support. + *

+ * Deriving assumes racks are not in use (see the class javadoc), whereas Cassandra's replica assignment is + * rack aware, so the derived ranges are only guaranteed to match for a single rack per datacenter. Supplying + * the ranges removes that assumption. + */ + private List explicitRanges; private transient RangeMap> replicas; private transient Multimap> tokenRangeMap; @@ -110,6 +137,26 @@ public CassandraRing(Partitioner partitioner, String keyspace, ReplicationFactor replicationFactor, Collection instances) + { + this(partitioner, keyspace, replicationFactor, instances, null); + } + + /** + * Constructs a ring from token ranges reported by Cassandra, instead of deriving them from tokens and the + * replication factor. + * + * @param partitioner the partitioner + * @param keyspace the keyspace this ring describes + * @param replicationFactor the keyspace replication factor + * @param instances all instances in the ring + * @param rangeReplicas range to replicas mapping as reported by Cassandra. Replicas must be present in + * {@code instances}. Ranges must be open-closed and non-wrapping. + */ + public CassandraRing(Partitioner partitioner, + String keyspace, + ReplicationFactor replicationFactor, + Collection instances, + Map, ? extends Collection> rangeReplicas) { this.partitioner = partitioner; this.keyspace = keyspace; @@ -117,41 +164,97 @@ public CassandraRing(Partitioner partitioner, this.instances = instances.stream() .sorted(Comparator.comparing(instance -> new BigInteger(instance.token()))) .collect(Collectors.toCollection(ArrayList::new)); + this.explicitRanges = rangeReplicas == null ? null : toExplicitRanges(rangeReplicas, this.instances); this.init(); } + /** + * Flattens the supplied mapping into a serializable form, resolving each replica to its index in + * {@code sortedInstances} so instance details are not duplicated per range. + */ + private static List toExplicitRanges( + Map, ? extends Collection> rangeReplicas, + List sortedInstances) + { + Map indexByInstance = new HashMap<>(sortedInstances.size()); + for (int index = 0; index < sortedInstances.size(); index++) + { + indexByInstance.put(sortedInstances.get(index), index); + } + + List result = new ArrayList<>(rangeReplicas.size()); + rangeReplicas.forEach((range, replicas) -> { + Preconditions.checkArgument(range.lowerEndpoint().compareTo(range.upperEndpoint()) <= 0, + "Supplied ranges must not wrap, found %s", range); + List indexes = new ArrayList<>(replicas.size()); + for (CassandraInstance replica : replicas) + { + Integer index = indexByInstance.get(replica); + Preconditions.checkArgument(index != null, + "Replica %s for range %s is not present in the ring instances", + replica, range); + indexes.add(index); + } + result.add(new ExplicitRange(range.lowerEndpoint(), range.upperEndpoint(), indexes)); + }); + result.sort(Comparator.comparing(r -> r.lower)); + return result; + } + + /** + * @return {@code true} if this ring uses token ranges reported by Cassandra rather than locally derived ones + */ + public boolean hasExplicitRanges() + { + return explicitRanges != null; + } + private void init() { // Setup token range map replicas = TreeRangeMap.create(); tokenRangeMap = ArrayListMultimap.create(); - // Calculate instance to token ranges mapping - switch (replicationFactor.getReplicationStrategy()) + if (explicitRanges != null) { - case SimpleStrategy: - tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(instances, - replicationFactor.getTotalReplicationFactor(), - partitioner)); - break; - case NetworkTopologyStrategy: - for (String dataCenter : dataCenters()) + for (ExplicitRange explicitRange : explicitRanges) + { + Range range = Range.openClosed(explicitRange.lower, explicitRange.upper); + for (int index : explicitRange.replicaIndexes) { - int rf = replicationFactor.getOptions().get(dataCenter); - if (rf == 0) + tokenRangeMap.put(instances.get(index), range); + } + } + } + else + { + // Calculate instance to token ranges mapping + switch (replicationFactor.getReplicationStrategy()) + { + case SimpleStrategy: + tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(instances, + replicationFactor.getTotalReplicationFactor(), + partitioner)); + break; + case NetworkTopologyStrategy: + for (String dataCenter : dataCenters()) { - continue; + int rf = replicationFactor.getOptions().get(dataCenter); + if (rf == 0) + { + continue; + } + List dcInstances = instances.stream() + .filter(instance -> instance.dataCenter().matches(dataCenter)) + .collect(Collectors.toList()); + tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(dcInstances, + replicationFactor.getOptions().get(dataCenter), + partitioner)); } - List dcInstances = instances.stream() - .filter(instance -> instance.dataCenter().matches(dataCenter)) - .collect(Collectors.toList()); - tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(dcInstances, - replicationFactor.getOptions().get(dataCenter), - partitioner)); - } - break; - default: - throw new UnsupportedOperationException("Unsupported replication strategy"); + break; + default: + throw new UnsupportedOperationException("Unsupported replication strategy"); + } } // Calculate token range to replica mapping @@ -159,6 +262,55 @@ private void init() tokenRangeMap.asMap().forEach((instance, ranges) -> ranges.forEach(range -> addReplica(instance, range, replicas))); } + /** + * A single Cassandra-reported token range and the instances replicating it, held by index into + * {@link CassandraRing#instances}. + */ + private static final class ExplicitRange implements Serializable + { + private static final long serialVersionUID = 2026090800000000001L; + + private final BigInteger lower; + private final BigInteger upper; + private final List replicaIndexes; + + private ExplicitRange(BigInteger lower, BigInteger upper, List replicaIndexes) + { + this.lower = lower; + this.upper = upper; + this.replicaIndexes = replicaIndexes; + } + + @Override + public boolean equals(Object other) + { + if (this == other) + { + return true; + } + if (other == null || getClass() != other.getClass()) + { + return false; + } + ExplicitRange that = (ExplicitRange) other; + return Objects.equals(lower, that.lower) + && Objects.equals(upper, that.upper) + && Objects.equals(replicaIndexes, that.replicaIndexes); + } + + @Override + public int hashCode() + { + return Objects.hash(lower, upper, replicaIndexes); + } + + @Override + public String toString() + { + return "(" + lower + ", " + upper + "]=" + replicaIndexes; + } + } + public Partitioner partitioner() { return partitioner; @@ -247,6 +399,7 @@ public boolean equals(Object other) && Objects.equals(this.keyspace, that.keyspace) && Objects.equals(this.replicationFactor, that.replicationFactor) && Objects.equals(this.instances, that.instances) + && Objects.equals(this.explicitRanges, that.explicitRanges) && Objects.equals(this.replicas, that.replicas) && Objects.equals(this.tokenRangeMap, that.tokenRangeMap); } @@ -254,12 +407,18 @@ public boolean equals(Object other) @Override public int hashCode() { - return Objects.hash(partitioner, keyspace, replicationFactor, instances, replicas, tokenRangeMap); + return Objects.hash(partitioner, keyspace, replicationFactor, instances, explicitRanges, replicas, tokenRangeMap); } private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { LOGGER.debug("Falling back to JDK deserialization"); + byte formatVersion = in.readByte(); + if (formatVersion != SERIALIZATION_FORMAT_VERSION) + { + throw new IOException(String.format("Unsupported CassandraRing serialization format version %d, expected %d", + formatVersion, SERIALIZATION_FORMAT_VERSION)); + } this.partitioner = in.readByte() == 0 ? Partitioner.RandomPartitioner : Partitioner.Murmur3Partitioner; this.keyspace = in.readUTF(); @@ -270,7 +429,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); @@ -278,12 +443,36 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE { this.instances.add(new CassandraInstance(in.readUTF(), in.readUTF(), in.readUTF())); } + + int numExplicitRanges = in.readInt(); + if (numExplicitRanges < 0) + { + this.explicitRanges = null; + } + else + { + List ranges = new ArrayList<>(numExplicitRanges); + for (int range = 0; range < numExplicitRanges; range++) + { + BigInteger lower = new BigInteger(in.readUTF()); + BigInteger upper = new BigInteger(in.readUTF()); + int numReplicas = in.readInt(); + List indexes = new ArrayList<>(numReplicas); + for (int replica = 0; replica < numReplicas; replica++) + { + indexes.add(in.readInt()); + } + ranges.add(new ExplicitRange(lower, upper, indexes)); + } + this.explicitRanges = ranges; + } this.init(); } private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFoundException { LOGGER.debug("Falling back to JDK serialization"); + out.writeByte(SERIALIZATION_FORMAT_VERSION); out.writeByte(this.partitioner == Partitioner.RandomPartitioner ? 0 : 1); out.writeUTF(this.keyspace); @@ -295,6 +484,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) @@ -303,6 +499,26 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou out.writeUTF(instance.nodeName()); out.writeUTF(instance.dataCenter()); } + + // -1 distinguishes "derive the ranges" from "explicitly zero ranges" + if (this.explicitRanges == null) + { + out.writeInt(-1); + } + else + { + out.writeInt(this.explicitRanges.size()); + for (ExplicitRange range : this.explicitRanges) + { + out.writeUTF(range.lower.toString()); + out.writeUTF(range.upper.toString()); + out.writeInt(range.replicaIndexes.size()); + for (int index : range.replicaIndexes) + { + out.writeInt(index); + } + } + } } public static class Serializer extends com.esotericsoftware.kryo.Serializer @@ -314,17 +530,61 @@ public void write(Kryo kryo, Output out, CassandraRing ring) out.writeString(ring.keyspace); kryo.writeObject(out, ring.replicationFactor); kryo.writeObject(out, ring.instances); + // -1 distinguishes "derive the ranges" from "explicitly zero ranges" + if (ring.explicitRanges == null) + { + out.writeInt(-1); + } + else + { + out.writeInt(ring.explicitRanges.size()); + for (ExplicitRange range : ring.explicitRanges) + { + out.writeString(range.lower.toString()); + out.writeString(range.upper.toString()); + out.writeInt(range.replicaIndexes.size()); + for (int index : range.replicaIndexes) + { + out.writeInt(index); + } + } + } } @Override @SuppressWarnings("unchecked") public CassandraRing read(Kryo kryo, Input in, Class type) { - return new CassandraRing(in.readByte() == 1 ? Partitioner.RandomPartitioner - : Partitioner.Murmur3Partitioner, - in.readString(), - kryo.readObject(in, ReplicationFactor.class), - kryo.readObject(in, ArrayList.class)); + Partitioner partitioner = in.readByte() == 1 ? Partitioner.RandomPartitioner + : Partitioner.Murmur3Partitioner; + String keyspace = in.readString(); + ReplicationFactor replicationFactor = kryo.readObject(in, ReplicationFactor.class); + List instances = kryo.readObject(in, ArrayList.class); + + int numExplicitRanges = in.readInt(); + if (numExplicitRanges < 0) + { + return new CassandraRing(partitioner, keyspace, replicationFactor, instances); + } + + // Instances are sorted by the constructor, and were written in that order, so indexes still resolve + List sorted = instances.stream() + .sorted(Comparator.comparing(i -> new BigInteger(i.token()))) + .collect(Collectors.toList()); + Map, List> rangeReplicas = new HashMap<>(numExplicitRanges); + for (int range = 0; range < numExplicitRanges; range++) + { + BigInteger lower = new BigInteger(in.readString()); + BigInteger upper = new BigInteger(in.readString()); + int numReplicas = in.readInt(); + List replicas = new ArrayList<>(numReplicas); + for (int replica = 0; replica < numReplicas; replica++) + { + replicas.add(sorted.get(in.readInt())); + } + rangeReplicas.put(Range.openClosed(lower, upper), replicas); + } + return new CassandraRing(partitioner, keyspace, replicationFactor, instances, rangeReplicas); } } } 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 9d989be51..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 @@ -52,6 +52,9 @@ public final class CqlUtils "max_index_interval" ); private static final Pattern REPLICATION_FACTOR_PATTERN = Pattern.compile("WITH REPLICATION = (\\{[^\\}]*\\})"); + private static final String TRACKED_REPLICATION_TYPE = "tracked"; + private static final Pattern REPLICATION_TYPE_PATTERN = Pattern.compile("replication_type\\s*=\\s*'(\\w+)'", + Pattern.CASE_INSENSITIVE); // Initialize a mapper allowing single quotes to process the RF string from the CREATE KEYSPACE statement private static final ObjectMapper MAPPER = new ObjectMapper().configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); private static final Pattern ESCAPED_WHITESPACE_PATTERN = Pattern.compile("(\\\\r|\\\\n|\\\\r\\n)+"); @@ -173,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) @@ -296,4 +308,33 @@ public static boolean isTimeRangeFilterSupported(String compactionStrategy) { return compactionStrategy == null || compactionStrategy.endsWith("TimeWindowCompactionStrategy"); } + + /** + * Extracts replication type from create schema statement + * + * @param schemaStr full cluster schema string as returned by Sidecar + * @param keyspace name of the keyspace to check + * @return {@code true} if keyspace is tracked {@code false} otherwise + */ + public static String extractReplicationType(@NotNull String schemaStr, @NotNull String keyspace) + { + String createKeyspaceSchema = extractKeyspaceSchema(schemaStr, keyspace); + Matcher matcher = REPLICATION_TYPE_PATTERN.matcher(createKeyspaceSchema); + if (matcher.find()) + { + return matcher.group(1); + } + return null; + } + + /** + * Returns {@code true} if {@code replication_type = 'tracked'} in create statement otherwise {@code false} + * + * @param replicationType replication type extracted from create statement + * @return {@code true} if replication type is tracked {@code false} otherwise + */ + public static boolean isTracked(String replicationType) + { + return TRACKED_REPLICATION_TYPE.equalsIgnoreCase(replicationType); + } } 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..d1d1ba412 100644 --- a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java +++ b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java @@ -19,8 +19,16 @@ package org.apache.cassandra.spark.data; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.lang.reflect.Field; import java.util.ArrayList; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; @@ -111,4 +119,325 @@ public void testEquality() assertThat(replicationFactor1).isEqualTo(replicationFactor2); assertThat(replicationFactor1.hashCode()).isEqualTo(replicationFactor2.hashCode()); } + + // Transient / witness replicas: the / form, reused by witness replicas under + // mutation tracking (CEP-45/CEP-46) + + @Test + public void testNoTransientReplicasByDefault() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3")); + assertThat(replicationFactor.hasTransientReplicas()).isFalse(); + assertThat(replicationFactor.getTransientOptions()).isEmpty(); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(0); + assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(3); + assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(0); + } + + @Test + public void testTransientReplicasSingleDatacenter() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThat(replicationFactor.hasTransientReplicas()).isTrue(); + // total keeps its original meaning: all replicas, witnesses included + assertThat(replicationFactor.getOptions().get("datacenter1")).isEqualTo(Integer.valueOf(3)); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(1); + assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(2); + assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1); + } + + @Test + public void testTransientReplicasMultipleDatacenters() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3/1")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(4); + assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(2); + } + + @Test + public void testMixedTransientAndFullDatacenters() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(5); + assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(replicationFactor.getTransientReplicas("datacenter2")).isEqualTo(0); + assertThat(replicationFactor.getFullReplicas("datacenter2")).isEqualTo(3); + // datacenter2 has no transient replicas, so it must be absent rather than mapped to zero + assertThat(replicationFactor.getTransientOptions()).containsOnlyKeys("datacenter1"); + } + + @Test + public void testTransientReplicasSimpleStrategy() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "SimpleStrategy", + "replication_factor", "3/1")); + assertThat(replicationFactor.getReplicationStrategy()) + .isEqualTo(ReplicationFactor.ReplicationStrategy.SimpleStrategy); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + } + + @Test + public void testTransientReplicasWithWhitespace() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", " 3 / 1 ")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + } + + @Test + public void testTransientEqualToTotalIsRejected() + { + // Cassandra requires at least one full replica, so 3/3 is invalid + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/3")); + assertThat(replicationFactor.getOptions()).doesNotContainKey("datacenter1"); + } + + @Test + public void testMalformedTransientValuesAreSkipped() + { + for (String malformed : new String[]{"3/", "/1", "3/1/1", "3/x", "x/1", "3/-1", ""}) + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", malformed)); + assertThat(replicationFactor.getOptions()) + .as("malformed value '%s' should not produce a replication factor entry", malformed) + .doesNotContainKey("datacenter1"); + } + } + + @Test + public void testGetFullReplicasUnknownDatacenter() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThatThrownBy(() -> replicationFactor.getFullReplicas("nosuchdc")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testTransientOptionsForUnknownDatacenterRejected() + { + assertThatThrownBy(() -> new ReplicationFactor( + ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy, + ImmutableMap.of("datacenter1", 3), + ImmutableMap.of("datacenter2", 1))) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testEqualityConsidersTransientReplicas() + { + ReplicationFactor full = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3")); + ReplicationFactor withTransient = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThat(full).isNotEqualTo(withTransient); + assertThat(full.hashCode()).isNotEqualTo(withTransient.hashCode()); + } + + @Test + public void testNegativeReplicationFactorIsRejected() + { + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "-3")); + assertThat(replicationFactor.getOptions()).doesNotContainKey("datacenter1"); + } + + @Test + public void testZeroReplicationFactorIsAllowed() + { + // RF 0 is legitimate for NetworkTopologyStrategy: the keyspace is simply not replicated to that datacenter + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3", + "datacenter2", "0")); + assertThat(replicationFactor.getOptions().get("datacenter2")).isEqualTo(Integer.valueOf(0)); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + } + + // parseStrict: same parsing, but an unparseable value raises instead of dropping the datacenter + + @Test + public void testParseStrictRaisesOnUnparseableValue() + { + assertThatThrownBy(() -> ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "xyz"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("datacenter1"); + } + + @Test + public void testParseStrictAcceptsTransientForm() + { + ReplicationFactor replicationFactor = ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1")); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3); + assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2); + } + + @Test + public void testLenientConstructorStillDropsUnparseableValue() + { + // The lenient constructor is retained for callers that tolerate a partial replication factor + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3", + "datacenter2", "xyz")); + assertThat(replicationFactor.getOptions()).containsOnlyKeys("datacenter1"); + } + + @Test + public void testParseStrictRaisesWhenNoDatacenterEntries() + { + assertThatThrownBy(() -> ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Could not find replication info in schema map"); + } + + @Test + public void testParseStrictRaisesWhenEveryDatacenterIsUnparseable() + { + // Every entry dropped is the same situation as no entries at all, and must not yield an empty + // replication factor that fails later with a misleading message + assertThatThrownBy(() -> ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "xyz"))) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testParseStrictAllowsLocalStrategyWithNoEntries() + { + // LocalStrategy legitimately has no datacenter entries, e.g. the system_schema keyspace + ReplicationFactor replicationFactor = ReplicationFactor.parseStrict(ImmutableMap.of( + "class", "org.apache.cassandra.locator.LocalStrategy")); + assertThat(replicationFactor.getReplicationStrategy()) + .isEqualTo(ReplicationFactor.ReplicationStrategy.LocalStrategy); + assertThat(replicationFactor.getOptions()).isEmpty(); + assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(0); + } + + @Test + public void testLenientConstructorAllowsNoDatacenterEntries() + { + // Unchanged lenient behaviour: no guard, so CDC callers are unaffected + ReplicationFactor replicationFactor = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy")); + assertThat(replicationFactor.getOptions()).isEmpty(); + } + + // Serialization: a new field that silently fails to round-trip would surface as wrong + // replication data on Spark executors, so cover both paths with a non-zero transient count + + @Test + public void testKryoSerializationRoundTripWithTransientReplicas() throws Exception + { + ReplicationFactor original = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + + Kryo kryo = new Kryo(); + kryo.register(ReplicationFactor.class, new ReplicationFactor.Serializer()); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (Output out = new Output(bytes)) + { + kryo.writeObject(out, original); + } + ReplicationFactor deserialized; + try (Input in = new Input(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = kryo.readObject(in, ReplicationFactor.class); + } + + assertThat(deserialized).isEqualTo(original); + assertThat(deserialized.getTotalReplicationFactor()).isEqualTo(6); + assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5); + assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1); + assertThat(deserialized.getTransientReplicas("datacenter2")).isEqualTo(0); + } + + @Test + public void testJdkSerializationRoundTripWithTransientReplicas() throws Exception + { + ReplicationFactor original = new ReplicationFactor(ImmutableMap.of( + "class", "NetworkTopologyStrategy", + "datacenter1", "3/1", + "datacenter2", "3")); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(original); + } + ReplicationFactor deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (ReplicationFactor) in.readObject(); + } + + assertThat(deserialized).isEqualTo(original); + assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5); + assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1); + } + + @Test + public void testReadResolveToleratesLegacyEmptyOptions() throws Exception + { + // An instance serialized before transient replica support deserializes with a null transientOptions, and + // the older lenient constructor allowed a non-LocalStrategy replication factor with no datacenter entries. + // readResolve must normalise it rather than turn it into a deserialization failure. + ReplicationFactor legacy = new ReplicationFactor(ImmutableMap.of("class", "NetworkTopologyStrategy")); + assertThat(legacy.getOptions()).isEmpty(); + + Field transientOptions = ReplicationFactor.class.getDeclaredField("transientOptions"); + transientOptions.setAccessible(true); + transientOptions.set(legacy, null); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(legacy); + } + ReplicationFactor deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (ReplicationFactor) in.readObject(); + } + + assertThat(deserialized.getOptions()).isEmpty(); + assertThat(deserialized.hasTransientReplicas()).isFalse(); + assertThat(deserialized.getTransientReplicationFactor()).isEqualTo(0); + } } diff --git a/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java b/cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java index c66290f0f..1c9e5dbd6 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,11 +19,23 @@ 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; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Multimap; import com.google.common.collect.Range; @@ -34,6 +46,7 @@ import org.apache.cassandra.spark.data.ReplicationFactor; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; @SuppressWarnings("UnstableApiUsage") public class CassandraRingTests @@ -447,4 +460,255 @@ public void testNetworkStrategyRF22() Partitioner.Murmur3Partitioner.minToken(), Partitioner.Murmur3Partitioner.maxToken())); } + + @Test + public void testJdkSerializationPreservesTransientReplicas() throws Exception + { + // CassandraRing hand-rolls readObject/writeObject and rebuilds ReplicationFactor from strategy plus + // options, so a transient (witness) count could silently vanish on the way to a Spark executor + CassandraRing ring = new CassandraRing( + Partitioner.Murmur3Partitioner, + "test", + new ReplicationFactor(ImmutableMap.of("class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "DC1", "3/1", + "DC2", "3")), + Arrays.asList(new CassandraInstance("0", "local0-i1", "DC1"), + new CassandraInstance("100", "local0-i2", "DC1"), + new CassandraInstance("200", "local0-i3", "DC1"), + new CassandraInstance("1", "local1-i1", "DC2"), + new CassandraInstance("101", "local1-i2", "DC2"), + new CassandraInstance("201", "local1-i3", "DC2"))); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(ring); + } + CassandraRing deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (CassandraRing) in.readObject(); + } + + ReplicationFactor rf = deserialized.replicationFactor(); + assertThat(rf.getTransientReplicas("DC1")).isEqualTo(1); + assertThat(rf.getTransientReplicas("DC2")).isEqualTo(0); + assertThat(rf.getTotalReplicationFactor()).isEqualTo(6); + assertThat(rf.getFullReplicationFactor()).isEqualTo(5); + assertThat(rf).isEqualTo(ring.replicationFactor()); + // Deliberately not asserting deserialized.equals(ring): CassandraRing#equals compares the derived + // replicas and tokenRangeMap fields, and does not hold across a JDK round trip even without transient + // replicas. Pre-existing behaviour, unrelated to replica types. + } + + @Test + public void testJdkDeserializationRejectsUnknownFormatVersion() throws Exception + { + // The hand-rolled format gained a transient-replica section, but the class signature did not change, so the + // computed serialVersionUID would still match an older stream. A leading format-version byte makes a stale + // stream fail loudly instead of being misread. + CassandraRing ring = new CassandraRing( + Partitioner.Murmur3Partitioner, + "test", + new ReplicationFactor(ImmutableMap.of("class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "DC1", "3")), + Arrays.asList(new CassandraInstance("0", "local0-i1", "DC1"), + new CassandraInstance("100", "local0-i2", "DC1"), + new CassandraInstance("200", "local0-i3", "DC1"))); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(ring); + } + + // Locate the format-version byte deterministically: the format is + // [version][partitioner] then writeUTF(keyspace), so it sits two bytes before the "test" UTF header + byte[] raw = bytes.toByteArray(); + byte[] keyspaceUtf = new byte[]{0, 4, 't', 'e', 's', 't'}; + int keyspaceIndex = indexOf(raw, keyspaceUtf); + assertThat(keyspaceIndex).as("keyspace UTF header should be locatable").isGreaterThan(1); + int versionIndex = keyspaceIndex - 2; + // Deliberately not asserting the current version number, so this test does not need updating + // every time the format changes + assertThat(raw[versionIndex]).as("format version byte should be positive").isGreaterThan((byte) 0); + + // Simulate a stream written by a version whose format we do not understand + raw[versionIndex] = (byte) 99; + + 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; + } + + // Cassandra-reported (explicit) token ranges, used for witness-enabled keyspaces where the locally + // derived ranges cannot be trusted because the derivation ignores racks + + private static CassandraRing explicitRangeRing() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1"), + new CassandraInstance("200", "n3", "dc1")); + // Deliberately not what the derivation would produce: boundary at 50, and only two of three + // replicas per range. A derived ring can never produce this shape. + Map, List> explicit = new LinkedHashMap<>(); + explicit.put(Range.openClosed(Partitioner.Murmur3Partitioner.minToken(), BigInteger.valueOf(50L)), + Arrays.asList(instances.get(0), instances.get(1))); + explicit.put(Range.openClosed(BigInteger.valueOf(50L), Partitioner.Murmur3Partitioner.maxToken()), + Arrays.asList(instances.get(1), instances.get(2))); + return new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "3/1")), + instances, explicit); + } + + private static Set replicaNamesAt(CassandraRing ring, long token) + { + Set names = new TreeSet<>(); + ring.getReplicas(BigInteger.valueOf(token)).forEach(instance -> names.add(instance.nodeName())); + return names; + } + + @Test + public void testExplicitRangesAreHonouredInsteadOfDerived() + { + CassandraRing explicitRing = explicitRangeRing(); + assertThat(explicitRing.hasExplicitRanges()).isTrue(); + // the supplied boundary at 50 is used, and only the supplied replicas are returned + assertThat(replicaNamesAt(explicitRing, 10L)).containsExactly("n1", "n2"); + assertThat(replicaNamesAt(explicitRing, 1000L)).containsExactly("n2", "n3"); + } + + @Test + public void testDerivedRingIsUnchangedByExplicitRangeSupport() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1"), + new CassandraInstance("200", "n3", "dc1")); + CassandraRing derived = new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "3")), + instances); + assertThat(derived.hasExplicitRanges()).isFalse(); + // derivation puts boundaries at node tokens, so 50 is not a boundary + Set upperBounds = new TreeSet<>(); + derived.rangeMap().asMapOfRanges().forEach((range, replicas) -> { + if (!replicas.isEmpty()) + { + upperBounds.add(range.upperEndpoint()); + } + }); + assertThat(upperBounds).doesNotContain(BigInteger.valueOf(50L)); + } + + @Test + public void testExplicitRangesSurviveJdkSerialization() throws Exception + { + CassandraRing original = explicitRangeRing(); + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) + { + out.writeObject(original); + } + CassandraRing deserialized; + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = (CassandraRing) in.readObject(); + } + assertThat(deserialized.hasExplicitRanges()).isTrue(); + assertThat(replicaNamesAt(deserialized, 10L)).containsExactly("n1", "n2"); + assertThat(replicaNamesAt(deserialized, 1000L)).containsExactly("n2", "n3"); + } + + @Test + public void testExplicitRangesSurviveKryoSerialization() + { + CassandraRing original = explicitRangeRing(); + Kryo kryo = new Kryo(); + kryo.register(CassandraRing.class, new CassandraRing.Serializer()); + kryo.register(ReplicationFactor.class, new ReplicationFactor.Serializer()); + kryo.register(CassandraInstance.class, new CassandraInstance.Serializer()); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (Output out = new Output(bytes)) + { + kryo.writeObject(out, original); + } + CassandraRing deserialized; + try (Input in = new Input(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = kryo.readObject(in, CassandraRing.class); + } + assertThat(deserialized.hasExplicitRanges()).isTrue(); + assertThat(replicaNamesAt(deserialized, 10L)).containsExactly("n1", "n2"); + } + + @Test + public void testDerivedRingStaysDerivedThroughKryo() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1"), + new CassandraInstance("200", "n3", "dc1")); + CassandraRing derived = new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "3")), + instances); + Kryo kryo = new Kryo(); + kryo.register(CassandraRing.class, new CassandraRing.Serializer()); + kryo.register(ReplicationFactor.class, new ReplicationFactor.Serializer()); + kryo.register(CassandraInstance.class, new CassandraInstance.Serializer()); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (Output out = new Output(bytes)) + { + kryo.writeObject(out, derived); + } + CassandraRing deserialized; + try (Input in = new Input(new ByteArrayInputStream(bytes.toByteArray()))) + { + deserialized = kryo.readObject(in, CassandraRing.class); + } + assertThat(deserialized.hasExplicitRanges()).isFalse(); + assertThat(deserialized).isEqualTo(derived); + } + + @Test + public void testExplicitRangeWithUnknownReplicaIsRejected() + { + List instances = Arrays.asList(new CassandraInstance("0", "n1", "dc1"), + new CassandraInstance("100", "n2", "dc1")); + Map, List> explicit = new LinkedHashMap<>(); + explicit.put(Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(100L)), + Arrays.asList(new CassandraInstance("999", "nope", "dc1"))); + assertThatThrownBy(() -> new CassandraRing(Partitioner.Murmur3Partitioner, "test", + new ReplicationFactor(ImmutableMap.of( + "class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "2")), + instances, explicit)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("not present in the ring instances"); + } } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java index a9c120871..53ac97974 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java @@ -356,6 +356,17 @@ public ReplicationFactor replicationFactor() } } + @Override + public String getReplicationType() + { + String keyspaceSchema = getKeyspaceSchema(true); + if (keyspaceSchema == null) + { + throw new RuntimeException("Could not retrieve keyspace schema information for keyspace " + conf.keyspace); + } + return CqlUtils.extractReplicationType(keyspaceSchema, conf.keyspace); + } + @Override public TokenRangeMapping getTokenRangeMapping(boolean cached) { diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java index bed36ac4b..fb82483a7 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContext.java @@ -46,13 +46,24 @@ public CassandraDirectDataTransportContext(@NotNull BulkWriterContext bulkWriter } @Override - public DirectStreamSession createStreamSession(BulkWriterContext writerContext, - String sessionId, - SortedSSTableWriter sstableWriter, - Range range, - ReplicaAwareFailureHandler failureHandler, - ExecutorService executorService) + public StreamSession createStreamSession( + BulkWriterContext writerContext, + String sessionId, + SortedSSTableWriter sstableWriter, + Range range, + ReplicaAwareFailureHandler failureHandler, + ExecutorService executorService) { + if (bridge().isTracked(clusterInfo.getReplicationType())) + { + return new TrackedDirectStreamSession(writerContext, + sstableWriter, + this, + sessionId, + range, + failureHandler, + executorService); + } return new DirectStreamSession(writerContext, sstableWriter, this, @@ -71,7 +82,11 @@ public DirectDataTransferApi dataTransferApi() // only invoke in constructor protected DirectDataTransferApi createDirectDataTransferApi() { - CassandraBridge bridge = CassandraBridgeFactory.get(clusterInfo.getLowestCassandraVersion()); - return new SidecarDataTransferApi(clusterInfo.getCassandraContext(), bridge, jobInfo); + return new SidecarDataTransferApi(clusterInfo.getCassandraContext(), bridge(), jobInfo); + } + + private CassandraBridge bridge() + { + return CassandraBridgeFactory.get(clusterInfo.getLowestCassandraVersion()); } } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java index 64b654449..e4d8a0a0b 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/ClusterInfo.java @@ -82,6 +82,16 @@ public interface ClusterInfo extends StartupValidatable */ ReplicationFactor replicationFactor(); + /** + * @return {@code replication_type} of the enclosing keyspace (e.g. {@code "tracked"}, {@code "untracked"}), + * or {@code null} if replication_type is absent + */ + @Nullable + default String getReplicationType() + { + return null; + } + CassandraContext getCassandraContext(); /** diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TrackedDirectStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TrackedDirectStreamSession.java new file mode 100644 index 000000000..166ef2387 --- /dev/null +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TrackedDirectStreamSession.java @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.cassandra.spark.bulkwriter; + +import java.math.BigInteger; +import java.util.Set; +import java.util.concurrent.ExecutorService; + +import com.google.common.collect.Range; + +import org.apache.cassandra.bridge.SSTableDescriptor; +import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler; + +/** + * Stream session for bulk writes to keyspaces with mutation tracking enabled. + * + *

+ * Tracked stream session uploads and triggers import on the coordinator node only. Cassandra's coordinated + * transfer then propagates the data to all other replicas, avoiding the duplicate-row updates that would occur if each + * replica independently streamed the data to its peers. + *

+ */ +public class TrackedDirectStreamSession extends StreamSession +{ + public TrackedDirectStreamSession(BulkWriterContext writerContext, + SortedSSTableWriter sstableWriter, + TransportContext.DirectDataBulkWriterContext transportContext, + String sessionID, + Range tokenRange, + ReplicaAwareFailureHandler failureHandler, + ExecutorService executorService) + { + super(writerContext, sstableWriter, transportContext, sessionID, tokenRange, failureHandler, executorService); + } + + @Override + protected void onSSTablesProduced(Set sstables) + { + throw new UnsupportedOperationException("TrackedDirectStreamSession is not yet implemented"); + } + + @Override + protected StreamResult doFinalizeStream() + { + throw new UnsupportedOperationException("TrackedDirectStreamSession is not yet implemented"); + } + + @Override + protected void sendRemainingSSTables() + { + throw new UnsupportedOperationException("TrackedDirectStreamSession is not yet implemented"); + } +} diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java index 99cbc6827..692f8d138 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java @@ -51,6 +51,8 @@ import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import com.google.common.collect.Range; +import com.google.common.collect.RangeSet; +import com.google.common.collect.TreeRangeSet; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -62,6 +64,7 @@ import o.a.c.sidecar.client.shaded.common.response.NodeSettings; import o.a.c.sidecar.client.shaded.common.response.RingResponse; import o.a.c.sidecar.client.shaded.common.response.SchemaResponse; +import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse; import org.apache.cassandra.analytics.stats.Stats; import org.apache.cassandra.bridge.BigNumberConfig; import org.apache.cassandra.bridge.BigNumberConfigImpl; @@ -317,7 +320,25 @@ private int initBulkReader(@NotNull ClientConfig options) throws ExecutionExcept udts.forEach(udt -> LOGGER.info("Adding schema UDT: '{}'", udt)); cqlTable = bridge().buildSchema(createStmt, keyspace, replicationFactor, partitioner, udts, null, indexCount, false); - CassandraRing ring = createCassandraRingFromRing(partitioner, replicationFactor, ringFuture.get()); + + // Mutation tracked keyspaces may replicate a range to a witness that holds no data, so the range to replica + // mapping must come from Cassandra rather than being derived locally. See + // createCassandraRingFromTokenRangeReplicas. + String replicationType = CqlUtils.extractReplicationType(fullSchema, keyspace); + boolean tracked = bridge().isTracked(replicationType); + CassandraRing ring; + if (tracked || options.forceCassandraTokenRanges()) + { + LOGGER.info("Sourcing token ranges from Cassandra keyspace={} replicationType={} forced={}", + keyspace, replicationType, options.forceCassandraTokenRanges()); + TokenRangeReplicasResponse topology = sidecar.tokenRangeReplicas(new ArrayList<>(clusterConfig), + maybeQuotedKeyspace).get(); + ring = createCassandraRingFromTokenRangeReplicas(partitioner, replicationFactor, ringFuture.get(), topology); + } + else + { + ring = createCassandraRingFromRing(partitioner, replicationFactor, ringFuture.get()); + } int effectiveNumberOfCores = sizingFuture.get(); tokenPartitioner = new TokenPartitioner(ring, options.defaultParallelism(), effectiveNumberOfCores); @@ -691,6 +712,123 @@ public CassandraRing createCassandraRingFromRing(Partitioner partitioner, return new CassandraRing(partitioner, keyspace, replicationFactor, instances); } + /** + * Builds the ring using the token ranges Cassandra reports, rather than deriving them from tokens and the + * replication factor. The local derivation assumes racks are not in use, whereas Cassandra's replica assignment + * is rack aware, so for mutation tracked keyspaces - where a replica may be a witness holding no data - the + * derived ranges cannot be relied on to identify which instance replicates which range. + *

+ * Node discovery still comes from the ring response, so snapshot creation, sizing and the Sidecar client pool + * are unaffected. Only the range to replica mapping is taken from Cassandra. + * + * @param partitioner the partitioner + * @param replicationFactor the keyspace replication factor + * @param ring the ring response, used for node discovery + * @param topology the token range replicas response, used for the range to replica mapping + * @return a ring whose token ranges were reported by Cassandra + */ + public CassandraRing createCassandraRingFromTokenRangeReplicas(Partitioner partitioner, + ReplicationFactor replicationFactor, + RingResponse ring, + TokenRangeReplicasResponse topology) + { + List instances = ring + .stream() + .filter(status -> datacenter == null || datacenter.equalsIgnoreCase(status.datacenter())) + .map(status -> new CassandraInstance(status.token(), status.fqdn(), status.datacenter())) + .collect(Collectors.toList()); + + // Token range replicas identify replicas by "address:port", while instances built from the ring are keyed + // by fqdn, so the replica metadata is used to translate between them + Map instanceByNodeName = new HashMap<>(instances.size()); + for (CassandraInstance instance : instances) + { + CassandraInstance previous = instanceByNodeName.put(instance.nodeName(), instance); + if (previous != null && !previous.equals(instance)) + { + // A node owning several tokens cannot be represented as a single CassandraInstance. Mutation + // tracking requires num_tokens=1, so this should not happen on a tracked keyspace's cluster + throw new IllegalStateException(String.format( + "Node %s owns multiple tokens (%s and %s). Sourcing token ranges from Cassandra requires " + + "single-token nodes.", instance.nodeName(), previous.token(), instance.token())); + } + } + + Map, List> rangeReplicas = new HashMap<>(); + for (TokenRangeReplicasResponse.ReplicaInfo replicaInfo : topology.readReplicas()) + { + Range range = Range.openClosed(new BigInteger(replicaInfo.start()), + new BigInteger(replicaInfo.end())); + List replicas = new ArrayList<>(); + for (Map.Entry> byDatacenter : replicaInfo.replicasByDatacenter().entrySet()) + { + if (datacenter != null && !datacenter.equalsIgnoreCase(byDatacenter.getKey())) + { + continue; + } + for (String replicaKey : byDatacenter.getValue()) + { + TokenRangeReplicasResponse.ReplicaMetadata metadata = topology.replicaMetadata().get(replicaKey); + String nodeName = metadata != null ? metadata.fqdn() : replicaKey; + CassandraInstance instance = instanceByNodeName.get(nodeName); + if (instance == null) + { + // The ring response is the source of truth for which nodes the reader can talk to, so a + // replica absent from it is skipped rather than failing the job + LOGGER.warn("Skipping replica reported by Cassandra that is absent from the ring " + + "replica={} fqdn={} range=({}, {}]", + replicaKey, nodeName, replicaInfo.start(), replicaInfo.end()); + continue; + } + replicas.add(instance); + } + } + if (!replicas.isEmpty()) + { + // Distinct, because a duplicated replica would inflate the per-range replica count that + // PartitionedDataLayer uses to decide whether the consistency level is satisfied + rangeReplicas.put(range, replicas.stream().distinct().collect(Collectors.toList())); + } + } + + if (rangeReplicas.isEmpty()) + { + throw new IllegalStateException( + "Cassandra reported no read replicas for keyspace " + keyspace + " in datacenter " + datacenter); + } + + validateCompleteRingCoverage(partitioner, rangeReplicas.keySet()); + + LOGGER.info("Using token ranges reported by Cassandra keyspace={} numRanges={} numInstances={}", + keyspace, rangeReplicas.size(), instances.size()); + return new CassandraRing(partitioner, keyspace, replicationFactor, instances, rangeReplicas); + } + + /** + * Fails if the supplied ranges leave any part of the token ring without a replica. + *

+ * A gap is not caught downstream. {@link CassandraRing} seeds its range map across the whole ring, so an + * uncovered range is present but with an empty replica list, and the reader only discovers this per Spark + * partition as a {@code NotEnoughReplicasException} that reads like a cluster availability problem. Reporting it + * here names the actual cause. + * + * @param partitioner the partitioner, for the ring bounds + * @param ranges the ranges reported by Cassandra that have at least one usable replica + */ + private void validateCompleteRingCoverage(Partitioner partitioner, Set> ranges) + { + RangeSet uncovered = TreeRangeSet.create(); + uncovered.add(Range.openClosed(partitioner.minToken(), partitioner.maxToken())); + ranges.forEach(uncovered::remove); + if (!uncovered.isEmpty()) + { + throw new IllegalStateException(String.format( + "Token ranges reported by Cassandra do not cover the whole ring for keyspace %s in datacenter %s. " + + "Uncovered: %s. This means some ranges have no replica the reader can use, which would otherwise " + + "surface later as a consistency failure.", keyspace, datacenter, uncovered)); + } + } + // Startup Validation @Override diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java index 4ca6c9214..89463da7f 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java @@ -55,6 +55,15 @@ public class ClientConfig public static final String SNAPSHOT_NAME_KEY = "snapshotName"; public static final String DC_KEY = "dc"; public static final String CREATE_SNAPSHOT_KEY = "createSnapshot"; + /** + * Forces token ranges to be sourced from Cassandra via the token-range-replicas endpoint even for keyspaces + * that are not mutation tracked. Normally this happens only for tracked keyspaces. Exists so the code path can + * be exercised against ordinary clusters, which is the only coverage available until Cassandra 6.0 bridge + * modules land and tracked keyspaces can be created in integration tests. + *

+ * Set as {@code forcecassandratokenranges}; Spark lowercases option keys. + */ + public static final String FORCE_CASSANDRA_TOKEN_RANGES_KEY = "forceCassandraTokenRanges"; public static final String CLEAR_SNAPSHOT_KEY = "clearSnapshot"; /** * Format of clearSnapshotStrategy is {strategy [snapshotTTLvalue]}, clearSnapshotStrategy holds both the strategy @@ -101,6 +110,7 @@ public class ClientConfig protected String snapshotName; protected String datacenter; protected boolean createSnapshot; + protected boolean forceCassandraTokenRanges; protected boolean clearSnapshot; protected ClearSnapshotStrategy clearSnapshotStrategy; protected int defaultParallelism; @@ -127,6 +137,7 @@ protected ClientConfig(Map options) this.snapshotName = MapUtils.getOrDefault(options, SNAPSHOT_NAME_KEY, "sbr_" + UUID.randomUUID().toString().replace("-", "")); this.datacenter = options.get(MapUtils.lowerCaseKey(DC_KEY)); this.createSnapshot = MapUtils.getBoolean(options, CREATE_SNAPSHOT_KEY, true); + this.forceCassandraTokenRanges = MapUtils.getBoolean(options, FORCE_CASSANDRA_TOKEN_RANGES_KEY, false); this.clearSnapshot = MapUtils.getBoolean(options, CLEAR_SNAPSHOT_KEY, createSnapshot); String clearSnapshotStrategyOption = MapUtils.getOrDefault(options, CLEAR_SNAPSHOT_STRATEGY_KEY, null); @@ -204,6 +215,14 @@ public String datacenter() return datacenter; } + /** + * @return {@code true} to source token ranges from Cassandra regardless of whether the keyspace is tracked + */ + public boolean forceCassandraTokenRanges() + { + return forceCassandraTokenRanges; + } + public boolean createSnapshot() { return createSnapshot; diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java index 5ed4c5221..763082a36 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java @@ -30,8 +30,10 @@ import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse; import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping; import org.apache.cassandra.spark.exception.TimeSkewTooLargeException; +import org.apache.cassandra.spark.utils.CqlUtils; import static org.apache.cassandra.spark.TestUtils.range; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; @@ -75,6 +77,50 @@ public static CassandraClusterInfo mockClusterInfoForTimeSkewTest(int allowanceM return new MockClusterInfoForTimeSkew(allowanceMinutes, remoteNow); } + @Test + void testGetTrackedReplicationType() + { + String schema = "CREATE KEYSPACE mykeyspace " + + "WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': '3'}" + + " AND durable_writes = true" + + " AND replication_type = 'tracked';"; + CassandraClusterInfo ci = mockClusterInfoWithSchema("mykeyspace", schema); + assertThat(ci.getReplicationType()) + .describedAs("Keyspace with replication_type = 'tracked' should return 'tracked'") + .isEqualTo("tracked"); + } + + @Test + void testGetUntrackedReplicationType() + { + String schema = "CREATE KEYSPACE k " + + "WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': '3'}" + + " AND durable_writes = true" + + " AND replication_type = 'untracked';"; + CassandraClusterInfo ci = mockClusterInfoWithSchema("k", schema); + assertThat(ci.getReplicationType()) + .describedAs("Keyspace with replication_type = 'untracked' should return 'untracked'") + .isEqualTo("untracked"); + } + + @Test + void testGetReplicationTypeReturnsNullWhenPropertyAbsent() + { + // Schema without replication_type — pre-mutation-tracking clusters + String schema = "CREATE KEYSPACE k " + + "WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': '3'}" + + " AND durable_writes = true;"; + CassandraClusterInfo ci = mockClusterInfoWithSchema("k", schema); + assertThat(ci.getReplicationType()) + .describedAs("Keyspace without replication_type property should return null") + .isNull(); + } + + private static CassandraClusterInfo mockClusterInfoWithSchema(String keyspace, String schemaStr) + { + return new MockClusterInfoForSchema(keyspace, schemaStr); + } + private static class MockClusterInfoForTimeSkew extends CassandraClusterInfo { private CassandraContext cassandraContext; @@ -107,4 +153,39 @@ private void mockCassandraContext(int allowanceMinutes, Instant remoteNow) when(cassandraContext.sidecarPort()).thenReturn(9043); } } + + /** + * Minimal ClusterInfo stub that returns a fixed keyspace schema string, used to test + * {@link CassandraClusterInfo#getReplicationType()}. + */ + private static class MockClusterInfoForSchema extends CassandraClusterInfo + { + private final String keyspaceName; + private final String schemaStr; + + MockClusterInfoForSchema(String keyspace, String schemaStr) + { + super((BulkSparkConf) null); + this.keyspaceName = keyspace; + this.schemaStr = schemaStr; + } + + @Override + protected CassandraContext buildCassandraContext() + { + return mock(CassandraContext.class, RETURNS_DEEP_STUBS); + } + + @Override + public String getKeyspaceSchema(boolean cached) + { + return schemaStr; + } + + @Override + public String getReplicationType() + { + return CqlUtils.extractReplicationType(schemaStr, keyspaceName); + } + } } diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContextTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContextTest.java new file mode 100644 index 000000000..0bd2170bb --- /dev/null +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraDirectDataTransportContextTest.java @@ -0,0 +1,151 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.cassandra.spark.bulkwriter; + +import java.math.BigInteger; +import java.nio.file.Path; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import com.google.common.collect.BoundType; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Range; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.apache.cassandra.spark.bulkwriter.token.MultiClusterReplicaAwareFailureHandler; +import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler; +import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping; +import org.apache.cassandra.spark.data.ReplicationFactor; +import org.apache.cassandra.spark.utils.XXHash32DigestAlgorithm; + +import static org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that {@link CassandraDirectDataTransportContext#createStreamSession}. + */ +public class CassandraDirectDataTransportContextTest +{ + @TempDir + private Path folder; + + private MockBulkWriterContext writerContext; + private MockTableWriter tableWriter; + private ExecutorService executor; + private Range range; + + @BeforeEach + public void setup() + { + ImmutableMap rfOptions = ImmutableMap.of("DC1", 3); + ReplicationFactor rf = new ReplicationFactor(NetworkTopologyStrategy, rfOptions); + TokenRangeMapping tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(0, rfOptions, 12); + writerContext = new MockBulkWriterContext(tokenRangeMapping); + writerContext.setReplicationFactor(rf); + tableWriter = new MockTableWriter(folder); + range = Range.range(BigInteger.valueOf(101L), BoundType.CLOSED, BigInteger.valueOf(199L), BoundType.CLOSED); + executor = Executors.newSingleThreadExecutor(); + } + + @AfterEach + public void tearDown() + { + executor.shutdownNow(); + } + + @Test + void createDirectStreamSessionForUntrackedKeyspace() + { + writerContext.setTrackedKeyspace(false); + CassandraDirectDataTransportContext transportContext = stubTransportContext(); + + StreamSession session = transportContext.createStreamSession( + writerContext, + "session-untracked", + new SortedSSTableWriter(tableWriter, folder, new XXHash32DigestAlgorithm(), 1), + range, + buildFailureHandler(), + executor); + + assertThat(session) + .describedAs("Untracked keyspace should produce a DirectStreamSession") + .isInstanceOf(DirectStreamSession.class); + } + + @Test + void createTrackedDirectStreamSessionForTrackedKeyspace() + { + writerContext.setTrackedKeyspace(true); + CassandraDirectDataTransportContext transportContext = stubTransportContext(); + + StreamSession session = transportContext.createStreamSession( + writerContext, + "session-tracked", + new SortedSSTableWriter(tableWriter, folder, new XXHash32DigestAlgorithm(), 1), + range, + buildFailureHandler(), + executor); + + assertThat(session) + .describedAs("Tracked keyspace should produce a TrackedDirectStreamSession") + .isInstanceOf(TrackedDirectStreamSession.class); + } + + @Test + void createDirectStreamSessionForAbsentReplicationType() + { + CassandraDirectDataTransportContext transportContext = stubTransportContext(); + + StreamSession session = transportContext.createStreamSession( + writerContext, + "session-null-type", + new SortedSSTableWriter(tableWriter, folder, new XXHash32DigestAlgorithm(), 1), + range, + buildFailureHandler(), + executor); + + assertThat(session) + .describedAs("Null replication type (pre-mutation-tracking) should produce a DirectStreamSession") + .isInstanceOf(DirectStreamSession.class); + } + + private CassandraDirectDataTransportContext stubTransportContext() + { + DirectDataTransferApi mockApi = + ((TransportContext.DirectDataBulkWriterContext) writerContext.transportContext()).dataTransferApi(); + + return new CassandraDirectDataTransportContext(writerContext) + { + @Override + protected DirectDataTransferApi createDirectDataTransferApi() + { + return mockApi; + } + }; + } + + private ReplicaAwareFailureHandler buildFailureHandler() + { + return new MultiClusterReplicaAwareFailureHandler<>(writerContext.cluster().getPartitioner()); + } +} diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java index cb5564f0c..be05715c4 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java @@ -99,6 +99,7 @@ public interface CommitResultSupplier extends BiFunction, String, D private final UUID jobId; private boolean skipClean = false; + private String replicationType = null; public int refreshClusterInfoCallCount = 0; // CHECKSTYLE IGNORE: Public mutable field private final Map> uploads = new ConcurrentHashMap<>(); private final Map> commits = new ConcurrentHashMap<>(); @@ -204,6 +205,17 @@ public void setReplicationFactor(ReplicationFactor replicationFactor) this.replicationFactor = replicationFactor; } + @Override + public String getReplicationType() + { + return replicationType; + } + + public void setTrackedKeyspace(boolean tracked) + { + this.replicationType = tracked ? "tracked" : "untracked"; + } + @Override public CassandraContext getCassandraContext() { diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java index 088de9070..5fff7a30c 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java @@ -19,15 +19,31 @@ package org.apache.cassandra.spark.data; +import java.math.BigInteger; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import o.a.c.sidecar.client.shaded.common.response.RingResponse; +import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse; +import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse.ReplicaInfo; +import o.a.c.sidecar.client.shaded.common.response.TokenRangeReplicasResponse.ReplicaMetadata; +import o.a.c.sidecar.client.shaded.common.response.data.RingEntry; +import org.apache.cassandra.clients.Sidecar; +import org.apache.cassandra.spark.data.partitioner.CassandraInstance; +import org.apache.cassandra.spark.data.partitioner.CassandraRing; +import org.apache.cassandra.spark.data.partitioner.Partitioner; + import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class CassandraDataLayerTests { @@ -64,4 +80,239 @@ void testClearSnapshotOptionSupport(Boolean clearSnapshot, String expectedClearS assertThat(clearSnapshotStrategy.hasTTL()).isEqualTo(expectedClearSnapshotStrategy.hasTTL()); assertThat(clearSnapshotStrategy.ttl()).isEqualTo(expectedClearSnapshotStrategy.ttl()); } + + // Sourcing token ranges from Cassandra, used for mutation tracked keyspaces where the locally derived + // ranges cannot identify which instance replicates which range + + private static final String KS = "big-data"; + + private static CassandraDataLayer dataLayer(String datacenter) + { + Map options = new HashMap<>(REQUIRED_CLIENT_CONFIG_OPTIONS); + if (datacenter != null) + { + options.put("dc", datacenter); + } + return new CassandraDataLayer(ClientConfig.create(options), Sidecar.ClientConfig.create(), null); + } + + private static RingResponse ringOf(String... fqdnAndTokenAndDc) + { + RingResponse ring = new RingResponse(); + for (int i = 0; i < fqdnAndTokenAndDc.length; i += 3) + { + ring.add(new RingEntry.Builder().fqdn(fqdnAndTokenAndDc[i]) + .token(fqdnAndTokenAndDc[i + 1]) + .datacenter(fqdnAndTokenAndDc[i + 2]) + .address(fqdnAndTokenAndDc[i]) + .port(9042) + .rack("rack1") + .status("UP") + .state("NORMAL") + .load("1") + .owns("1") + .hostId("h" + i) + .build()); + } + return ring; + } + + /** + * Replica keys are "address:port" and are translated to fqdn through the replica metadata, mirroring what + * Sidecar returns. + */ + private static TokenRangeReplicasResponse topologyOf(List readReplicas, String... addressToFqdn) + { + Map metadata = new HashMap<>(); + for (int i = 0; i < addressToFqdn.length; i += 2) + { + String key = addressToFqdn[i]; + String fqdn = addressToFqdn[i + 1]; + metadata.put(key, new ReplicaMetadata("NORMAL", "UP", fqdn, key.split(":")[0], 9042, "dc1")); + } + return new TokenRangeReplicasResponse(readReplicas, readReplicas, metadata); + } + + @Test + void testRingFromTokenRangeReplicasUsesCassandraReportedRanges() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1", "n3", "200", "dc1"); + + // boundary at 50 with only two replicas: a shape the local derivation cannot produce + List readReplicas = Arrays.asList( + new ReplicaInfo("-9223372036854775808", "50", ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042"))), + new ReplicaInfo("50", "9223372036854775807", ImmutableMap.of("dc1", Arrays.asList("1.1.1.2:9042", "1.1.1.3:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2", + "1.1.1.3:9042", "n3"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(3), ring, topology); + + assertThat(result.hasExplicitRanges()).isTrue(); + assertThat(replicaNames(result, 10L)).containsExactly("n1", "n2"); + assertThat(replicaNames(result, 1000L)).containsExactly("n2", "n3"); + } + + @Test + void testRingFromTokenRangeReplicasFiltersByDatacenter() + { + CassandraDataLayer layer = dataLayer("dc1"); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1", "n4", "300", "dc2"); + + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042"), + "dc2", Collections.singletonList("1.1.1.4:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2", + "1.1.1.4:9042", "n4"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology); + + // the dc2 replica must not appear, even though Cassandra reported it + assertThat(replicaNames(result, 10L)).containsExactly("n1", "n2"); + } + + @Test + void testRingFromTokenRangeReplicasSkipsReplicaAbsentFromRing() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.9:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.9:9042", "ghost"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology); + + // the ring response is the source of truth for reachable nodes, so the unknown replica is dropped + assertThat(replicaNames(result, 10L)).containsExactly("n1"); + } + + @Test + void testRingFromTokenRangeReplicasRejectsMultiTokenNode() + { + CassandraDataLayer layer = dataLayer(null); + // same fqdn twice with different tokens, i.e. vnodes + RingResponse ring = ringOf("n1", "0", "dc1", "n1", "100", "dc1"); + + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Collections.singletonList("1.1.1.1:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, "1.1.1.1:9042", "n1"); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(1), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("owns multiple tokens"); + } + + @Test + void testRingFromTokenRangeReplicasFailsWhenNoReadReplicas() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1"); + TokenRangeReplicasResponse topology = topologyOf(Collections.emptyList()); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(1), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("no read replicas"); + } + + @Test + void testForceCassandraTokenRangesDefaultsToFalse() + { + assertThat(ClientConfig.create(new HashMap<>(REQUIRED_CLIENT_CONFIG_OPTIONS)) + .forceCassandraTokenRanges()).isFalse(); + + // Spark lowercases option keys, and MapUtils looks them up lowercased, so this is how a user sets it + Map options = new HashMap<>(REQUIRED_CLIENT_CONFIG_OPTIONS); + options.put("forcecassandratokenranges", "true"); + assertThat(ClientConfig.create(options).forceCassandraTokenRanges()).isTrue(); + } + + private static List replicaNames(CassandraRing ring, long token) + { + return ring.getReplicas(BigInteger.valueOf(token)) + .stream() + .map(CassandraInstance::nodeName) + .sorted() + .collect(Collectors.toList()); + } + + @Test + void testRingFromTokenRangeReplicasFailsWhenRingIsNotFullyCovered() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + // only the lower half of the ring is reported + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "0", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2"); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("do not cover the whole ring"); + } + + @Test + void testRingFromTokenRangeReplicasFailsWhenAllReplicasForARangeAreSkipped() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + // the upper range's only replica is absent from the ring, so that range ends up with no usable + // replica and must be reported as a coverage gap rather than silently dropped + List readReplicas = Arrays.asList( + new ReplicaInfo("-9223372036854775808", "0", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.2:9042"))), + new ReplicaInfo("0", "9223372036854775807", + ImmutableMap.of("dc1", Collections.singletonList("1.1.1.9:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2", + "1.1.1.9:9042", "ghost"); + + assertThatThrownBy(() -> layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("do not cover the whole ring"); + } + + @Test + void testRingFromTokenRangeReplicasDeduplicatesReplicas() + { + CassandraDataLayer layer = dataLayer(null); + RingResponse ring = ringOf("n1", "0", "dc1", "n2", "100", "dc1"); + + // the same replica reported twice must not be counted twice, since PartitionedDataLayer uses the + // per-range replica count to decide whether the consistency level is satisfied + List readReplicas = Collections.singletonList( + new ReplicaInfo("-9223372036854775808", "9223372036854775807", + ImmutableMap.of("dc1", Arrays.asList("1.1.1.1:9042", "1.1.1.1:9042", "1.1.1.2:9042")))); + TokenRangeReplicasResponse topology = topologyOf(readReplicas, + "1.1.1.1:9042", "n1", + "1.1.1.2:9042", "n2"); + + CassandraRing result = layer.createCassandraRingFromTokenRangeReplicas( + Partitioner.Murmur3Partitioner, ReplicationFactor.simpleStrategy(2), ring, topology); + + assertThat(result.getReplicas(BigInteger.valueOf(10L))).hasSize(2); + assertThat(replicaNames(result, 10L)).containsExactly("n1", "n2"); + } } 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 5bb363d97..ccf44d7e5 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) @@ -833,6 +911,43 @@ public void testTimeRangeFilterSupported() assertThat(isTimeRangeFilterSupported("")).isFalse(); } + @Test + public void testExtractReplicationType() + { + String schemaStrTracked + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true AND replication_type = 'tracked';"; + assertThat(CqlUtils.extractReplicationType(schemaStrTracked, "k")).isEqualTo("tracked"); + String schemaStrTrackedCaseInsensitive + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true AND REPLICATION_TYPE = 'Tracked';"; + assertThat(CqlUtils.extractReplicationType(schemaStrTrackedCaseInsensitive, "k")).isEqualTo("Tracked"); + String schemaStrUntracked + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true AND replication_type = 'untracked';"; + assertThat(CqlUtils.extractReplicationType(schemaStrUntracked, "k")).isEqualTo("untracked"); + String schemaStrNoReplicationType + = "CREATE KEYSPACE k WITH REPLICATION " + + "= { 'class' : 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3' }" + + " AND DURABLE_WRITES = true;"; + assertThat(CqlUtils.extractReplicationType(schemaStrNoReplicationType, "k")).isNull(); + } + + @Test + public void testIsTracked() + { + assertThat(CqlUtils.isTracked("tracked")).isTrue(); + assertThat(CqlUtils.isTracked("TRACKED")).isTrue(); + assertThat(CqlUtils.isTracked("Tracked")).isTrue(); + assertThat(CqlUtils.isTracked("untracked")).isFalse(); + assertThat(CqlUtils.isTracked("random")).isFalse(); + assertThat(CqlUtils.isTracked(null)).isFalse(); + assertThat(CqlUtils.isTracked("")).isFalse(); + } + private static String loadFullSchemaSample() throws IOException { Path fullSchemaSampleFile = ResourceUtils.writeResourceToPath(CqlUtilsTest.class.getClassLoader(), tempPath, "cql/fullSchema.cql"); diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java index 93a877ea3..585c49a7d 100644 --- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java +++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/distributed/impl/CassandraCluster.java @@ -137,19 +137,18 @@ public AbstractCluster initializeCluster(String versionString, clusterBuilder.withTokenSupplier(tokenSupplier) .withConfig(instanceConfigUpdater); - if (dcCount > 1) + // An explicit supplier is always honoured, including for a single datacenter, so that a test can place + // nodes of one datacenter in different racks. The built-in default only makes sense for multiple datacenters. + if (configuration.dcAndRackSupplier != null) { - if (configuration.dcAndRackSupplier != null) - { - clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, configuration.dcAndRackSupplier)); - } - else - { - clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, - (nodeId) -> nodeId % 2 != 0 ? - dcAndRack("datacenter1", "rack1") : - dcAndRack("datacenter2", "rack2"))); - } + clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, configuration.dcAndRackSupplier)); + } + else if (dcCount > 1) + { + clusterBuilder.withNodeIdTopology(networkTopology(finalNodeCount, + (nodeId) -> nodeId % 2 != 0 ? + dcAndRack("datacenter1", "rack1") : + dcAndRack("datacenter2", "rack2"))); } if (configuration.instanceInitializer != null) diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderCassandraTokenRangesTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderCassandraTokenRangesTest.java new file mode 100644 index 000000000..d81529a98 --- /dev/null +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderCassandraTokenRangesTest.java @@ -0,0 +1,154 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.analytics; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.sidecar.testing.QualifiedName; +import org.apache.cassandra.testing.ClusterBuilderConfiguration; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; + +import static org.apache.cassandra.testing.TestUtils.DC1_RF3; +import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE; +import static org.apache.cassandra.testing.TestUtils.uniqueTestTableFullName; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests bulk reads that source token ranges from Cassandra via the token-range-replicas endpoint, rather than + * deriving them locally from tokens and the replication factor. + *

+ * Mutation tracked keyspaces take that path automatically, because a range may be replicated to a witness holding + * no data and the local derivation cannot say which instance replicates which range. Tracked keyspaces cannot be + * created until Cassandra 6.0 bridge modules land (CASSANALYTICS-192), so these tests use the + * {@code forcecassandratokenranges} option to exercise the same path against an ordinary keyspace. + *

+ * Reads must return exactly the same data either way, so each test asserts the full dataset rather than only a row + * count - an incomplete read is the failure mode this path exists to prevent. + */ +class BulkReaderCassandraTokenRangesTest extends SharedClusterSparkIntegrationTestBase +{ + static final List DATASET = Arrays.asList("a", "b", "c", "d", "e", "f", "g"); + + QualifiedName table = uniqueTestTableFullName(TEST_KEYSPACE); + + @Override + protected ClusterBuilderConfiguration testClusterConfiguration() + { + return super.testClusterConfiguration() + .nodesPerDc(3); + } + + @Test + void testReadWithCassandraSuppliedTokenRanges() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load(); + assertCompleteDataset(data); + } + + /** + * The derived and Cassandra-supplied paths must agree. This is the check that would catch the ranges being + * misaligned, which for a witness-enabled keyspace would show up as silently missing rows. + */ + @Test + void testCassandraSuppliedRangesMatchDerivedRanges() + { + Dataset derived = bulkReaderDataFrame(table).load(); + Dataset supplied = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load(); + + assertCompleteDataset(derived); + assertCompleteDataset(supplied); + assertThat(sortedNames(supplied)) + .describedAs("reads with Cassandra-supplied ranges must return the same rows as derived ranges") + .isEqualTo(sortedNames(derived)); + } + + @Test + void testReadWithCassandraSuppliedTokenRangesAtQuorum() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .option("consistencyLevel", "LOCAL_QUORUM") + .load(); + assertCompleteDataset(data); + } + + /** + * Pushdown filters are applied against the Spark partition token ranges, which are derived from the ring's + * ranges, so they are worth exercising on the new path. + */ + @Test + void testPushDownFilterWithCassandraSuppliedTokenRanges() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load() + .filter("id = 3"); + + List rows = data.collectAsList(); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getInt(0)).isEqualTo(3); + assertThat(rows.get(0).getString(1)).isEqualTo(DATASET.get(3)); + } + + private void assertCompleteDataset(Dataset data) + { + List rows = data.collectAsList().stream() + .sorted(Comparator.comparing(row -> row.getInt(0))) + .collect(Collectors.toList()); + assertThat(rows).hasSize(DATASET.size()); + for (int i = 0; i < DATASET.size(); i++) + { + assertThat(rows.get(i).getInt(0)).isEqualTo(i); + assertThat(rows.get(i).getString(1)).isEqualTo(DATASET.get(i)); + } + } + + private List sortedNames(Dataset data) + { + return data.collectAsList().stream() + .map(row -> row.getInt(0) + "=" + row.getString(1)) + .sorted() + .collect(Collectors.toList()); + } + + @Override + protected void initializeSchemaForTest() + { + createTestKeyspace(TEST_KEYSPACE, DC1_RF3); + createTestTable(table, "CREATE TABLE IF NOT EXISTS %s (id int PRIMARY KEY, name text);"); + + IInstance firstRunningInstance = cluster.getFirstRunningInstance(); + for (int i = 0; i < DATASET.size(); i++) + { + firstRunningInstance.coordinator() + .execute(String.format("INSERT INTO %s (id, name) VALUES (%d, '%s');", + table, i, DATASET.get(i)), + ConsistencyLevel.ALL); + } + } +} diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java new file mode 100644 index 000000000..6e409a95b --- /dev/null +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java @@ -0,0 +1,172 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.analytics; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.sidecar.testing.QualifiedName; +import org.apache.cassandra.testing.ClusterBuilderConfiguration; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; + +import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack; +import static org.apache.cassandra.testing.TestUtils.DC1_RF3; +import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE; +import static org.apache.cassandra.testing.TestUtils.uniqueTestTableFullName; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests bulk reads against a datacenter split across several racks. + *

+ * This closes a coverage gap. Every other integration test places all nodes of a datacenter in a single rack, where + * Cassandra's rack-aware replica assignment reduces to "take the next RF nodes in ring order" - exactly what the + * reader derives locally. With one rack per replica, {@code NetworkTopologyStrategy} instead skips nodes to spread + * replicas across racks (see {@code acceptableRackRepeats}), so a derived replica set can disagree with Cassandra's. + *

+ * {@code CassandraRing} states that it assumes racks are not in use, so a divergence here would previously have been + * invisible in CI. + *

+ * Note what these tests can and cannot show. Every replica here is a full replica holding data, so both replica + * selections return all the rows and the reads agree. They are therefore a regression guard on the range mapping and + * the replica join, not a demonstration that witness replicas are handled: selecting the wrong replicas only loses + * data once some of them hold none, which requires a mutation tracked keyspace and so cannot be tested until + * Cassandra 6.0 bridge modules land (CASSANALYTICS-192). + */ +class BulkReaderMultiRackTest extends SharedClusterSparkIntegrationTestBase +{ + static final List DATASET = Arrays.asList("a", "b", "c", "d", "e", "f", "g"); + + QualifiedName table = uniqueTestTableFullName(TEST_KEYSPACE); + + @Override + protected ClusterBuilderConfiguration testClusterConfiguration() + { + // Four nodes, three racks, with the first two nodes sharing a rack, and RF 3. + // + // NetworkTopologyStrategy computes acceptableRackRepeats = RF - rackCount, which is 0 here, so a node whose + // rack has already been used is skipped. Tokens are handed out in node order, so for the range whose first + // replica is node 1 the reader derives the next three nodes in ring order, 1,2,3, while Cassandra must skip + // node 2 (rack1 again) and pick 1,3,4. The two topology sources therefore genuinely disagree, which is the + // situation no existing test creates. + // + // Four nodes is the minimum that produces a disagreement. With one node per rack, or with racks assigned + // round-robin, rack-aware selection coincides with ring order and the test would prove nothing. + return super.testClusterConfiguration() + .nodesPerDc(4) + .dcAndRackSupplier((nodeId) -> { + switch (nodeId) + { + case 1: + case 2: + return dcAndRack("datacenter1", "rack1"); + case 3: + return dcAndRack("datacenter1", "rack2"); + case 4: + return dcAndRack("datacenter1", "rack3"); + default: + return dcAndRack("", ""); + } + }); + } + + @Test + void testReadWithDerivedRangesOnMultiRackCluster() + { + assertCompleteDataset(bulkReaderDataFrame(table).load()); + } + + @Test + void testReadWithCassandraSuppliedRangesOnMultiRackCluster() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .load(); + assertCompleteDataset(data); + } + + /** + * The point of the test class: on a rack-aware cluster the two topology sources must still agree on the data + * returned. If they diverge, taking ranges from Cassandra is the only correct option for witness keyspaces. + */ + @Test + void testBothTopologySourcesAgreeOnMultiRackCluster() + { + List derived = sortedNames(bulkReaderDataFrame(table).load()); + List supplied = sortedNames(bulkReaderDataFrame(table, + Collections.singletonMap("forcecassandratokenranges", "true")) + .load()); + assertThat(supplied) + .describedAs("derived and Cassandra-supplied ranges must return the same rows on a rack-aware cluster") + .isEqualTo(derived); + assertThat(derived).hasSize(DATASET.size()); + } + + @Test + void testReadAtLocalQuorumOnMultiRackCluster() + { + Dataset data = bulkReaderDataFrame(table, Collections.singletonMap("forcecassandratokenranges", "true")) + .option("consistencyLevel", "LOCAL_QUORUM") + .load(); + assertCompleteDataset(data); + } + + private void assertCompleteDataset(Dataset data) + { + List rows = data.collectAsList().stream() + .sorted(Comparator.comparing(row -> row.getInt(0))) + .collect(Collectors.toList()); + assertThat(rows).hasSize(DATASET.size()); + for (int i = 0; i < DATASET.size(); i++) + { + assertThat(rows.get(i).getInt(0)).isEqualTo(i); + assertThat(rows.get(i).getString(1)).isEqualTo(DATASET.get(i)); + } + } + + private List sortedNames(Dataset data) + { + return data.collectAsList().stream() + .map(row -> row.getInt(0) + "=" + row.getString(1)) + .sorted() + .collect(Collectors.toList()); + } + + @Override + protected void initializeSchemaForTest() + { + createTestKeyspace(TEST_KEYSPACE, DC1_RF3); + createTestTable(table, "CREATE TABLE IF NOT EXISTS %s (id int PRIMARY KEY, name text);"); + + IInstance firstRunningInstance = cluster.getFirstRunningInstance(); + for (int i = 0; i < DATASET.size(); i++) + { + firstRunningInstance.coordinator() + .execute(String.format("INSERT INTO %s (id, name) VALUES (%d, '%s');", + table, i, DATASET.get(i)), + ConsistencyLevel.ALL); + } + } +} diff --git a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java index 67c5bc7a0..d2d67ab92 100644 --- a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java +++ b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridge.java @@ -62,6 +62,7 @@ import org.apache.cassandra.spark.sparksql.filters.SparkRangeFilter; import org.apache.cassandra.analytics.stats.Stats; import org.apache.cassandra.spark.sparksql.filters.SSTableTimeRangeFilter; +import org.apache.cassandra.spark.utils.CqlUtils; import org.apache.cassandra.spark.utils.TimeProvider; import org.apache.cassandra.util.CompressionUtil; import org.jetbrains.annotations.NotNull; @@ -167,6 +168,18 @@ public String maybeQuoteIdentifier(String identifier) return cassandraTypes().maybeQuoteIdentifier(identifier); } + /** + * @return {@code true} if mutation tracking is enabled given the replication type, {@code false} otherwise + */ + public boolean isTracked(String replicationType) + { + if (replicationType == null) + { + return false; + } + return CqlUtils.isTracked(replicationType); + } + // CQL Type Parsing public CqlField.CqlType readType(CqlField.CqlType.InternalType type, Input input) 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;