This is an automated email from the ASF dual-hosted git repository.
sarankk pushed a commit to branch mutation-tracking-support
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/mutation-tracking-support by
this push:
new 56261ab9 CASSANALYTICS-197: Source token ranges from Cassandra for
mutation tracked keyspaces (#244)
56261ab9 is described below
commit 56261ab98c25e312beedbe5aa0e22d66f84c23cb
Author: Mansi Khara <[email protected]>
AuthorDate: Thu Oct 29 10:15:03 2026 -0700
CASSANALYTICS-197: Source token ranges from Cassandra for mutation tracked
keyspaces (#244)
patch by Mansi Khara; reviewed by Saranya Krishnakumar, Jyothsna Konisa,
Arjun Ashok for CASSANALYTICS-197
---
CHANGES.txt | 1 +
.../spark/data/partitioner/CassandraRing.java | 240 ++++++++++++++++---
.../spark/data/partitioner/CassandraRingTests.java | 247 ++++++++++++++++++-
.../cassandra/spark/data/CassandraDataLayer.java | 140 ++++++++++-
.../apache/cassandra/spark/data/ClientConfig.java | 12 +
.../spark/data/CassandraDataLayerTests.java | 252 ++++++++++++++++++++
.../distributed/impl/CassandraCluster.java | 23 +-
.../BulkReaderCassandraTokenRangesTest.java | 154 ++++++++++++
.../analytics/BulkReaderMultiRackTest.java | 263 +++++++++++++++++++++
9 files changed, 1278 insertions(+), 54 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 16f9ff38..4d503cbc 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Source token ranges from Cassandra for keyspaces with transient replicas
(CASSANALYTICS-197)
* Update 6.0 dtest version to SHA from cep-45-mutation-tracking branch
(CASSANALYTICS-200)
* Parse <replicas>/<transient> replication factor for witness-enabled
keyspaces (CASSANALYTICS-194)
* Determine whether mutation tracking is enabled for keyspace for bulk writes
(CASSANALYTICS-160)
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 5f12e1a8..563babcc 100644
---
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
+++
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
@@ -28,7 +28,10 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.Objects;
@@ -56,7 +59,10 @@ import static
org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrat
* 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.
* <p>
- * 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 keyspace with witness replicas, supply the ranges
instead - see
+ * {@link #CassandraRing(Partitioner, String, ReplicationFactor, Collection,
Map)}.
* <p>
* {@link #equals(Object)} and {@link #hashCode()} don't take {@link
#replicas} and {@link #tokenRangeMap}
* into consideration as they are just derived fields.
@@ -77,16 +83,23 @@ public class CassandraRing implements Serializable
/**
* Incremented whenever the hand-rolled JDK serialization format below
changes. Version 1 writes the
- * ReplicationFactor as an object rather than destructuring it.
+ * ReplicationFactor as an object rather than destructuring it. Version 2
added the optional
+ * Cassandra-reported token ranges after the instances: each instance
followed by the ranges it replicates.
*/
- private static final byte SERIALIZATION_FORMAT_VERSION = 1;
+ private static final byte SERIALIZATION_FORMAT_VERSION = 2;
private Partitioner partitioner;
private String keyspace;
private ReplicationFactor replicationFactor;
private List<CassandraInstance> instances;
+ /**
+ * Whether {@link #tokenRangeMap} was supplied by Cassandra rather than
derived locally from tokens and the
+ * replication factor. Deriving assumes racks are not in use; Cassandra
supplying the ranges removes that assumption.
+ */
+ private boolean rangesReportedByCassandra;
private transient RangeMap<BigInteger, List<CassandraInstance>> replicas;
+ // Each instance to the token ranges it replicates: derived in init(),
unless reported by Cassandra
private transient Multimap<CassandraInstance, Range<BigInteger>>
tokenRangeMap;
/**
@@ -122,53 +135,141 @@ public class CassandraRing implements Serializable
String keyspace,
ReplicationFactor replicationFactor,
Collection<CassandraInstance> instances)
+ {
+ this(partitioner, keyspace, replicationFactor, instances, null);
+ }
+
+ /**
+ * Constructs a ring from Cassandra-reported token ranges, or derives them
if none are supplied.
+ *
+ * @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, or {@code null} to derive it.
+ * Replicas must be present in {@code instances}.
+ */
+ public CassandraRing(Partitioner partitioner,
+ String keyspace,
+ ReplicationFactor replicationFactor,
+ Collection<CassandraInstance> instances,
+ Map<Range<BigInteger>, ? extends
Collection<CassandraInstance>> rangeReplicas)
+ {
+ this.partitioner = partitioner;
+ this.keyspace = keyspace;
+ this.replicationFactor = replicationFactor;
+ this.instances = sortedDistinct(instances);
+ this.rangesReportedByCassandra = rangeReplicas != null;
+ this.tokenRangeMap = rangeReplicas == null ? null :
reportedTokenRanges(rangeReplicas, this.instances);
+ this.init();
+ }
+
+ /**
+ * Used by deserialization, which already holds the instances in order and
the instance to ranges mapping
+ */
+ private CassandraRing(Partitioner partitioner,
+ String keyspace,
+ ReplicationFactor replicationFactor,
+ List<CassandraInstance> instances,
+ Multimap<CassandraInstance, Range<BigInteger>>
reportedTokenRanges)
{
this.partitioner = partitioner;
this.keyspace = keyspace;
this.replicationFactor = replicationFactor;
- this.instances = instances.stream()
- .sorted(Comparator.comparing(instance -> new
BigInteger(instance.token())))
-
.collect(Collectors.toCollection(ArrayList::new));
+ this.instances = sortedDistinct(instances);
+ this.rangesReportedByCassandra = true;
+ this.tokenRangeMap = reportedTokenRanges;
this.init();
}
+ private static List<CassandraInstance>
sortedDistinct(Collection<CassandraInstance> instances)
+ {
+ return instances.stream()
+ .distinct()
+ .sorted(Comparator.comparing(instance -> new
BigInteger(instance.token())))
+ .collect(Collectors.toCollection(ArrayList::new));
+ }
+
+ /**
+ * Inverts the reported range to replicas mapping into the instance to
ranges form held by
+ * {@link #tokenRangeMap}, adding ranges in token order so that a ring and
its deserialized copy compare equal.
+ * Ranges are normalised to open-closed.
+ */
+ private static Multimap<CassandraInstance, Range<BigInteger>>
reportedTokenRanges(
+ Map<Range<BigInteger>, ? extends Collection<CassandraInstance>>
rangeReplicas,
+ List<CassandraInstance> instances)
+ {
+ Set<CassandraInstance> ringInstances = new HashSet<>(instances);
+ Multimap<CassandraInstance, Range<BigInteger>> tokenRanges =
ArrayListMultimap.create();
+ rangeReplicas.entrySet()
+ .stream()
+ .sorted(Comparator.comparing(entry ->
entry.getKey().lowerEndpoint()))
+ .forEach(entry -> {
+ Range<BigInteger> range = entry.getKey();
+ Range<BigInteger> openClosed =
Range.openClosed(range.lowerEndpoint(), range.upperEndpoint());
+ for (CassandraInstance replica : entry.getValue())
+ {
+
Preconditions.checkArgument(ringInstances.contains(replica),
+ "Replica %s for range
%s is not present in the ring instances",
+ replica, range);
+ tokenRanges.put(replica, openClosed);
+ }
+ });
+ return tokenRanges;
+ }
+
+ /**
+ * @return {@code true} if this ring uses token ranges reported by
Cassandra rather than locally derived ones
+ */
+ public boolean hasReportedRanges()
+ {
+ return rangesReportedByCassandra;
+ }
+
private void init()
{
// Setup token range map
replicas = TreeRangeMap.create();
- tokenRangeMap = ArrayListMultimap.create();
- // Calculate instance to token ranges mapping
- switch (replicationFactor.getReplicationStrategy())
+ if (!rangesReportedByCassandra)
{
- case SimpleStrategy:
- tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(instances,
-
replicationFactor.getTotalReplicationFactor(),
-
partitioner));
- break;
- case NetworkTopologyStrategy:
- for (String dataCenter : dataCenters())
- {
- int rf = replicationFactor.getOptions().get(dataCenter);
- if (rf == 0)
+ // Calculate instance to token ranges mapping
+ tokenRangeMap = ArrayListMultimap.create();
+ 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<CassandraInstance> dcInstances =
instances.stream()
+ .filter(instance ->
instance.dataCenter().matches(dataCenter))
+ .collect(Collectors.toList());
+
tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(dcInstances,
+
replicationFactor.getOptions().get(dataCenter),
+
partitioner));
}
- List<CassandraInstance> 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
+ // Calculate token range to replica mapping. Replicas are added in
instance (token) order rather than the
+ // multimap's hash order, so each range lists its replicas in the same
order however the ring was built.
replicas.put(Range.openClosed(partitioner.minToken(),
partitioner.maxToken()), Collections.emptyList());
- tokenRangeMap.asMap().forEach((instance, ranges) ->
ranges.forEach(range -> addReplica(instance, range, replicas)));
+ for (CassandraInstance instance : instances)
+ {
+ tokenRangeMap.get(instance).forEach(range -> addReplica(instance,
range, replicas));
+ }
}
public Partitioner partitioner()
@@ -259,6 +360,7 @@ public class CassandraRing implements Serializable
&& Objects.equals(this.keyspace, that.keyspace)
&& Objects.equals(this.replicationFactor,
that.replicationFactor)
&& Objects.equals(this.instances, that.instances)
+ && this.rangesReportedByCassandra ==
that.rangesReportedByCassandra
&& Objects.equals(this.replicas, that.replicas)
&& Objects.equals(this.tokenRangeMap, that.tokenRangeMap);
}
@@ -266,7 +368,7 @@ public class CassandraRing implements Serializable
@Override
public int hashCode()
{
- return Objects.hash(partitioner, keyspace, replicationFactor,
instances, replicas, tokenRangeMap);
+ return Objects.hash(partitioner, keyspace, replicationFactor,
instances, rangesReportedByCassandra, replicas, tokenRangeMap);
}
private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException
@@ -291,6 +393,21 @@ public class CassandraRing implements Serializable
{
this.instances.add(new CassandraInstance(in.readUTF(),
in.readUTF(), in.readUTF()));
}
+
+ this.rangesReportedByCassandra = in.readBoolean();
+ if (rangesReportedByCassandra)
+ {
+ // Written as each instance followed by the ranges it replicates
+ this.tokenRangeMap = ArrayListMultimap.create();
+ for (CassandraInstance instance : this.instances)
+ {
+ int numRanges = in.readInt();
+ for (int range = 0; range < numRanges; range++)
+ {
+ tokenRangeMap.put(instance, Range.openClosed(new
BigInteger(in.readUTF()), new BigInteger(in.readUTF())));
+ }
+ }
+ }
this.init();
}
@@ -310,6 +427,21 @@ public class CassandraRing implements Serializable
out.writeUTF(instance.nodeName());
out.writeUTF(instance.dataCenter());
}
+
+ out.writeBoolean(this.rangesReportedByCassandra);
+ if (this.rangesReportedByCassandra)
+ {
+ for (CassandraInstance instance : this.instances)
+ {
+ Collection<Range<BigInteger>> ranges =
this.tokenRangeMap.get(instance);
+ out.writeInt(ranges.size());
+ for (Range<BigInteger> range : ranges)
+ {
+ out.writeUTF(range.lowerEndpoint().toString());
+ out.writeUTF(range.upperEndpoint().toString());
+ }
+ }
+ }
}
public static class Serializer extends
com.esotericsoftware.kryo.Serializer<CassandraRing>
@@ -321,17 +453,49 @@ public class CassandraRing implements Serializable
out.writeString(ring.keyspace);
kryo.writeObject(out, ring.replicationFactor);
kryo.writeObject(out, ring.instances);
+ out.writeBoolean(ring.rangesReportedByCassandra);
+ if (ring.rangesReportedByCassandra)
+ {
+ // Each instance followed by the ranges it replicates
+ for (CassandraInstance instance : ring.instances)
+ {
+ Collection<Range<BigInteger>> ranges =
ring.tokenRangeMap.get(instance);
+ out.writeInt(ranges.size());
+ for (Range<BigInteger> range : ranges)
+ {
+ out.writeString(range.lowerEndpoint().toString());
+ out.writeString(range.upperEndpoint().toString());
+ }
+ }
+ }
}
@Override
@SuppressWarnings("unchecked")
public CassandraRing read(Kryo kryo, Input in, Class<CassandraRing>
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<CassandraInstance> instances = kryo.readObject(in,
ArrayList.class);
+
+ if (!in.readBoolean())
+ {
+ return new CassandraRing(partitioner, keyspace,
replicationFactor, instances);
+ }
+
+ // Instances were written in order, each followed by the ranges it
replicates
+ Multimap<CassandraInstance, Range<BigInteger>> tokenRanges =
ArrayListMultimap.create();
+ for (CassandraInstance instance : instances)
+ {
+ int numRanges = in.readInt();
+ for (int range = 0; range < numRanges; range++)
+ {
+ tokenRanges.put(instance, Range.openClosed(new
BigInteger(in.readString()), new BigInteger(in.readString())));
+ }
+ }
+ return new CassandraRing(partitioner, keyspace, replicationFactor,
instances, tokenRanges);
}
}
}
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 a34e499d..dd525d9c 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
@@ -25,10 +25,19 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.math.BigInteger;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
+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;
@@ -522,10 +531,12 @@ public class CassandraRingTests
int keyspaceIndex = indexOf(raw, keyspaceUtf);
assertThat(keyspaceIndex).as("keyspace UTF header should be
locatable").isGreaterThan(1);
int versionIndex = keyspaceIndex - 2;
- assertThat(raw[versionIndex]).as("format version
byte").isEqualTo((byte) 1);
+ // 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 before the transient-replica section
existed
- raw[versionIndex] = 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)))
{
@@ -551,4 +562,234 @@ public class CassandraRingTests
}
return -1;
}
+
+ // Cassandra-reported token ranges, used for witness-enabled keyspaces
where the locally
+ // derived ranges cannot be trusted because the derivation ignores racks
+
+ private static CassandraRing reportedRangeRing()
+ {
+ List<CassandraInstance> 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<Range<BigInteger>, List<CassandraInstance>> reported = new
LinkedHashMap<>();
+
reported.put(Range.openClosed(Partitioner.Murmur3Partitioner.minToken(),
BigInteger.valueOf(50L)),
+ Arrays.asList(instances.get(0), instances.get(1)));
+ reported.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, reported);
+ }
+
+ private static Set<String> replicaNamesAt(CassandraRing ring, long token)
+ {
+ Set<String> names = new TreeSet<>();
+ ring.getReplicas(BigInteger.valueOf(token)).forEach(instance ->
names.add(instance.nodeName()));
+ return names;
+ }
+
+ @Test
+ public void testReportedRangesAreHonouredInsteadOfDerived()
+ {
+ CassandraRing reportedRing = reportedRangeRing();
+ assertThat(reportedRing.hasReportedRanges()).isTrue();
+ // the supplied boundary at 50 is used, and only the supplied replicas
are returned
+ assertThat(replicaNamesAt(reportedRing, 10L)).containsExactly("n1",
"n2");
+ assertThat(replicaNamesAt(reportedRing, 1000L)).containsExactly("n2",
"n3");
+ }
+
+ @Test
+ public void testDerivedRingIsUnchangedByReportedRangeSupport()
+ {
+ List<CassandraInstance> 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.hasReportedRanges()).isFalse();
+ // derivation puts boundaries at node tokens, so 50 is not a boundary
+ Set<BigInteger> 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 testReportedRangesSurviveJdkSerialization() throws Exception
+ {
+ CassandraRing original = reportedRangeRing();
+ 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.hasReportedRanges()).isTrue();
+ assertThat(replicaNamesAt(deserialized, 10L)).containsExactly("n1",
"n2");
+ assertThat(replicaNamesAt(deserialized, 1000L)).containsExactly("n2",
"n3");
+ }
+
+ @Test
+ public void testReportedRangesSurviveKryoSerialization()
+ {
+ CassandraRing original = reportedRangeRing();
+ 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.hasReportedRanges()).isTrue();
+ assertThat(replicaNamesAt(deserialized, 10L)).containsExactly("n1",
"n2");
+ assertThat(deserialized).isEqualTo(original);
+ }
+
+ @Test
+ public void testDerivedRingStaysDerivedThroughKryo()
+ {
+ List<CassandraInstance> 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.hasReportedRanges()).isFalse();
+ assertThat(deserialized).isEqualTo(derived);
+ }
+
+ @Test
+ public void testReportedRangeWithUnknownReplicaIsRejected()
+ {
+ List<CassandraInstance> instances = Arrays.asList(new
CassandraInstance("0", "n1", "dc1"),
+ new
CassandraInstance("100", "n2", "dc1"));
+ Map<Range<BigInteger>, List<CassandraInstance>> reported = new
LinkedHashMap<>();
+ reported.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, reported))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("not present in the ring instances");
+ }
+
+ private static CassandraRing reportedRing(Map<Range<BigInteger>,
List<CassandraInstance>> reported,
+ List<CassandraInstance>
instances)
+ {
+ return new CassandraRing(Partitioner.Murmur3Partitioner, "test",
+ new ReplicationFactor(ImmutableMap.of(
+ "class",
"org.apache.cassandra.locator.NetworkTopologyStrategy",
+ "dc1", "3")),
+ instances, reported);
+ }
+
+ @Test
+ public void testReportedRingDoesNotDependOnReplicaOrder()
+ {
+ // Sidecar lists replicas in ring-walk order, while a deserialized
ring lists them in instance order, so the
+ // ring must not depend on the order replicas were supplied in
+ List<CassandraInstance> instances = new ArrayList<>();
+ for (int node = 0; node < 32; node++)
+ {
+ instances.add(new CassandraInstance(String.valueOf(node * 100),
"n" + node, "dc1"));
+ }
+ Map<Range<BigInteger>, List<CassandraInstance>> forward = new
LinkedHashMap<>();
+ Map<Range<BigInteger>, List<CassandraInstance>> backward = new
LinkedHashMap<>();
+ for (int node = 0; node < instances.size(); node++)
+ {
+ BigInteger lower = node == 0 ?
Partitioner.Murmur3Partitioner.minToken() : BigInteger.valueOf((node - 1) *
100L + 1);
+ BigInteger upper = node == instances.size() - 1 ?
Partitioner.Murmur3Partitioner.maxToken() : BigInteger.valueOf(node * 100L);
+ List<CassandraInstance> replicas =
Arrays.asList(instances.get(node),
+
instances.get((node + 1) % instances.size()),
+
instances.get((node + 2) % instances.size()));
+ List<CassandraInstance> reversed = new ArrayList<>(replicas);
+ Collections.reverse(reversed);
+ forward.put(Range.openClosed(lower, upper), replicas);
+ backward.put(Range.openClosed(lower, upper), reversed);
+ }
+
+ CassandraRing ring = reportedRing(forward, instances);
+ CassandraRing reversedRing = reportedRing(backward, instances);
+ assertThat(reversedRing).isEqualTo(ring);
+ // whatever order they were supplied in, each range lists its replicas
in token order
+ reversedRing.rangeMap().asMapOfRanges().values().forEach(replicas -> {
+ List<CassandraInstance> sorted = new ArrayList<>(replicas);
+ sorted.sort((left, right) -> new
BigInteger(left.token()).compareTo(new BigInteger(right.token())));
+ assertThat(replicas).containsExactlyElementsOf(sorted);
+ });
+
+ 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, ring);
+ }
+ try (Input in = new Input(new
ByteArrayInputStream(bytes.toByteArray())))
+ {
+ assertThat(kryo.readObject(in,
CassandraRing.class)).isEqualTo(ring);
+ }
+ }
+
+ @Test
+ public void testReportedRangesAreNormalisedToOpenClosed()
+ {
+ List<CassandraInstance> instances = Arrays.asList(new
CassandraInstance("0", "n1", "dc1"),
+ new
CassandraInstance("100", "n2", "dc1"));
+ Map<Range<BigInteger>, List<CassandraInstance>> reported = new
LinkedHashMap<>();
+ reported.put(Range.closed(Partitioner.Murmur3Partitioner.minToken(),
BigInteger.ZERO),
+ Arrays.asList(instances.get(0), instances.get(1)));
+ reported.put(Range.openClosed(BigInteger.ZERO,
Partitioner.Murmur3Partitioner.maxToken()),
+ Arrays.asList(instances.get(0), instances.get(1)));
+
+ CassandraRing ring = reportedRing(reported, instances);
+
+ assertThat(ring.tokenRanges().get(instances.get(0)))
+
.containsExactly(Range.openClosed(Partitioner.Murmur3Partitioner.minToken(),
BigInteger.ZERO),
+ Range.openClosed(BigInteger.ZERO,
Partitioner.Murmur3Partitioner.maxToken()));
+ }
}
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 ff7e0b67..9a7b29c5 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.base.Preconditions;
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.ListSnapshotFilesResponse;
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;
@@ -327,7 +330,22 @@ public class CassandraDataLayer extends
PartitionedDataLayer implements StartupV
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());
+
+ // Keyspaces with transient replicas take their ranges from Cassandra;
see createCassandraRingFromTokenRangeReplicas
+ boolean hasWitnesses = replicationFactor.hasTransientReplicas();
+ CassandraRing ring;
+ if (hasWitnesses || options.forceCassandraTokenRanges())
+ {
+ LOGGER.info("Sourcing token ranges from Cassandra keyspace={}
transientReplicas={} forced={}",
+ keyspace, hasWitnesses,
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);
@@ -782,6 +800,126 @@ public class CassandraDataLayer extends
PartitionedDataLayer implements StartupV
return new CassandraRing(partitioner, keyspace, replicationFactor,
instances);
}
+ /**
+ * Builds the ring using the token ranges and replicas Cassandra reports,
rather than deriving them from tokens and
+ * the replication factor. There are two independent reasons to prefer
Cassandra's: the local derivation assumes
+ * racks are not in use, whereas Cassandra's replica assignment is rack
aware, and for keyspaces with transient
+ * replicas the reader needs Cassandra's exact placement to tell which
replica of a range is a witness. Only
+ * keyspaces with transient replicas use it by default, as they are
guaranteed single-token, which this path
+ * requires.
+ * <p>
+ * 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<CassandraInstance> instances = ring
+ .stream()
+ .filter(status -> datacenter ==
null || datacenter.equalsIgnoreCase(status.datacenter()))
+ .map(status -> new
CassandraInstance(status.token(), status.fqdn(), status.datacenter()))
+ .collect(Collectors.toList());
+
+ Map<String, CassandraInstance> 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. Cassandra
+ // rejects transient replicas when num_tokens > 1, so this is
only reachable when the path is forced
+ 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<Range<BigInteger>, List<CassandraInstance>> rangeReplicas = new
HashMap<>();
+ for (TokenRangeReplicasResponse.ReplicaInfo replicaInfo :
topology.readReplicas())
+ {
+ Range<BigInteger> range = Range.openClosed(new
BigInteger(replicaInfo.start()),
+ new
BigInteger(replicaInfo.end()));
+ List<CassandraInstance> replicas = new ArrayList<>();
+ for (Map.Entry<String, List<String>> byDatacenter :
replicaInfo.replicasByDatacenter().entrySet())
+ {
+ if (datacenter != null &&
!datacenter.equalsIgnoreCase(byDatacenter.getKey()))
+ {
+ continue;
+ }
+ for (String replicaKey : byDatacenter.getValue())
+ {
+ // Replicas are identified by "address:port", while
instances built from the ring are keyed by
+ // fqdn, so the replica metadata translates between them
+ 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. It should only happen while
+ // a cluster operation is in progress.
+ LOGGER.error("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.
+ * <p>
+ * 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<Range<BigInteger>> ranges)
+ {
+ RangeSet<BigInteger> 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 4ca6c921..2121d79b 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,8 @@ 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";
+ // Sources token ranges from Cassandra for any keyspace, not only those
with transient replicas
+ 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 +103,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 +130,7 @@ public class ClientConfig
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 +208,14 @@ public class ClientConfig
return datacenter;
}
+ /**
+ * @return {@code true} to source token ranges from Cassandra regardless
of whether the keyspace has witnesses
+ */
+ public boolean forceCassandraTokenRanges()
+ {
+ return forceCassandraTokenRanges;
+ }
+
public boolean createSnapshot()
{
return createSnapshot;
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 088de907..a4d5721a 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,240 @@ class CassandraDataLayerTests
assertThat(clearSnapshotStrategy.hasTTL()).isEqualTo(expectedClearSnapshotStrategy.hasTTL());
assertThat(clearSnapshotStrategy.ttl()).isEqualTo(expectedClearSnapshotStrategy.ttl());
}
+
+ // Sourcing token ranges from Cassandra, used for keyspaces with witness
replicas 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<String, String> 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<ReplicaInfo>
readReplicas, String... addressToFqdn)
+ {
+ Map<String, ReplicaMetadata> 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"));
+ }
+ // only the read replicas are used
+ return new TokenRangeReplicasResponse(Collections.emptyList(),
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<ReplicaInfo> 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.hasReportedRanges()).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<ReplicaInfo> 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<ReplicaInfo> 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<ReplicaInfo> 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<String, String> options = new
HashMap<>(REQUIRED_CLIENT_CONFIG_OPTIONS);
+ options.put("forcecassandratokenranges", "true");
+
assertThat(ClientConfig.create(options).forceCassandraTokenRanges()).isTrue();
+ }
+
+ private static List<String> 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<ReplicaInfo> 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<ReplicaInfo> 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<ReplicaInfo> 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-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 69686b90..85f8e0f4 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 class CassandraCluster<I extends IInstance>
implements IClusterExtension<
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 00000000..8416c72d
--- /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.
+ * <p>
+ * Keyspaces with transient replicas 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. Witness keyspaces
+ * cannot yet be created in these tests, so they use the
+ * {@code forcecassandratokenranges} option to exercise the same path against
an ordinary keyspace.
+ * <p>
+ * 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<String> 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<Row> 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<Row> derived = bulkReaderDataFrame(table).load();
+ Dataset<Row> 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<Row> 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<Row> data = bulkReaderDataFrame(table,
Collections.singletonMap("forcecassandratokenranges", "true"))
+ .load()
+ .filter("id = 3");
+
+ List<Row> 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<Row> data)
+ {
+ List<Row> 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<String> sortedNames(Dataset<Row> 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 00000000..75397277
--- /dev/null
+++
b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkReaderMultiRackTest.java
@@ -0,0 +1,263 @@
+/*
+ * 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.math.BigInteger;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.UUID;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+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.distributed.api.NodeToolResult;
+import org.apache.cassandra.sidecar.testing.QualifiedName;
+import org.apache.cassandra.spark.data.CassandraDataLayer;
+import org.apache.cassandra.spark.data.CassandraDataSourceHelper;
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.data.partitioner.CassandraRing;
+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.
+ * <p>
+ * 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.
+ * <p>
+ * {@code CassandraRing} states that it assumes racks are not in use, so a
divergence here would previously have been
+ * invisible in CI.
+ * <p>
+ * The reads cannot tell the two topology sources apart: every replica here
holds the data, so even a wrong replica
+ * set returns every row. {@link
#testCassandraSuppliedReplicasMatchCassandra()} therefore compares the reader's
rings
+ * with Cassandra's own placement directly. Reads from keyspaces where a wrong
replica holds no data, that is with
+ * witness replicas, are covered by the witness replica tests.
+ */
+class BulkReaderMultiRackTest extends SharedClusterSparkIntegrationTestBase
+{
+ static final List<String> DATASET = Arrays.asList("a", "b", "c", "d", "e",
"f", "g");
+ private static final Pattern DESCRIBE_RING_RANGE =
Pattern.compile("end_token:(-?\\d+), endpoints:\\[([^\\]]*)\\]");
+
+ 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<Row> data = bulkReaderDataFrame(table,
Collections.singletonMap("forcecassandratokenranges", "true"))
+ .load();
+ assertCompleteDataset(data);
+ }
+
+ /**
+ * Both topology sources return the same rows, as every replica holds the
data; see
+ * {@link #testCassandraSuppliedReplicasMatchCassandra()} for the check
that tells them apart.
+ */
+ @Test
+ void testBothTopologySourcesAgreeOnMultiRackCluster()
+ {
+ List<String> derived = sortedNames(bulkReaderDataFrame(table).load());
+ List<String> 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<Row> data = bulkReaderDataFrame(table,
Collections.singletonMap("forcecassandratokenranges", "true"))
+ .option("consistencyLevel", "LOCAL_QUORUM")
+ .load();
+ assertCompleteDataset(data);
+ }
+
+ /**
+ * The point of the test class. Taking replicas from Cassandra must
reproduce Cassandra's rack-aware placement for
+ * every range, while the local derivation, which ignores racks, must get
at least one range wrong on this cluster.
+ * The second assertion proves the cluster really exercises the
difference, so the first is not trivially true.
+ */
+ @Test
+ void testCassandraSuppliedReplicasMatchCassandra() throws
UnknownHostException
+ {
+ Map<BigInteger, Set<String>> cassandraPlacement =
cassandraReplicasByRangeEnd();
+ assertThat(cassandraPlacement).isNotEmpty();
+ CassandraRing supplied = readerRing(true);
+ CassandraRing derived = readerRing(false);
+
+ boolean derivedDiverges = false;
+ for (Map.Entry<BigInteger, Set<String>> range :
cassandraPlacement.entrySet())
+ {
+ assertThat(replicaNames(supplied, range.getKey()))
+ .describedAs("Cassandra-supplied replicas of the range ending at
%s", range.getKey())
+ .isEqualTo(range.getValue());
+ derivedDiverges |= !replicaNames(derived,
range.getKey()).equals(range.getValue());
+ }
+ assertThat(derivedDiverges)
+ .describedAs("locally derived replicas should differ from Cassandra's
for some range on a rack-aware cluster, "
+ + "otherwise this cluster does not exercise rack-aware
placement")
+ .isTrue();
+ }
+
+ /**
+ * Builds the ring a bulk read would use, without taking a snapshot
+ */
+ private CassandraRing readerRing(boolean forceCassandraTokenRanges)
+ {
+ getOrCreateSparkSession();
+ Map<String, String> options = new
HashMap<>(mtlsTestHelper.mtlOptionMap());
+ options.put("sidecar_contact_points",
SparkTestUtils.sidecarInstancesOptionStream(cluster, dnsResolver)
+
.collect(Collectors.joining(",")));
+ options.put("sidecar_port", String.valueOf(server.actualPort()));
+ options.put("keyspace", table.keyspace());
+ options.put("table", table.table());
+ options.put("DC", "datacenter1");
+ options.put("snapshotName", UUID.randomUUID().toString());
+ options.put("createSnapshot", "false");
+ options.put("clearSnapshotStrategy", "noop");
+ options.put("forcecassandratokenranges",
String.valueOf(forceCassandraTokenRanges));
+ CassandraDataLayer dataLayer = (CassandraDataLayer)
CassandraDataSourceHelper.getDataLayer(options, CassandraDataLayer::initialize);
+
assertThat(dataLayer.ring().hasReportedRanges()).isEqualTo(forceCassandraTokenRanges);
+ return dataLayer.ring();
+ }
+
+ private static Set<String> replicaNames(CassandraRing ring, BigInteger
token)
+ {
+ return
ring.getReplicas(token).stream().map(CassandraInstance::nodeName).collect(Collectors.toCollection(TreeSet::new));
+ }
+
+ /**
+ * Cassandra's own placement, from {@code nodetool describering}, keyed by
the inclusive end of each range
+ */
+ private Map<BigInteger, Set<String>> cassandraReplicasByRangeEnd() throws
UnknownHostException
+ {
+ NodeToolResult result =
cluster.getFirstRunningInstance().nodetoolResult("describering",
table.keyspace());
+ result.asserts().success();
+ Map<BigInteger, Set<String>> placement = new LinkedHashMap<>();
+ Matcher matcher = DESCRIBE_RING_RANGE.matcher(result.getStdout());
+ while (matcher.find())
+ {
+ Set<String> names = new TreeSet<>();
+ for (String endpoint : matcher.group(2).split(","))
+ {
+ String address = endpoint.trim().replaceFirst("^/",
"").replaceFirst(":\\d+$", "");
+ names.add(dnsResolver.reverseResolve(address));
+ }
+ placement.put(new BigInteger(matcher.group(1)), names);
+ }
+ return placement;
+ }
+
+ private void assertCompleteDataset(Dataset<Row> data)
+ {
+ List<Row> 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<String> sortedNames(Dataset<Row> 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);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
