mansikhara commented on code in PR #257:
URL:
https://github.com/apache/cassandra-analytics/pull/257#discussion_r4380893667
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java:
##########
@@ -782,6 +804,204 @@ 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 keyspaces with transient replicas - where a
replica may be a witness holding no data -
+ * the derived ranges cannot be relied on to identify which instance
replicates which range.
+ * <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());
+
+ // 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<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<>();
+ Map<Range<BigInteger>, List<CassandraInstance>> rangeWitnesses = new
HashMap<>();
+ for (TokenRangeReplicasResponse.ReplicaInfo replicaInfo :
topology.readReplicas())
+ {
+ Range<BigInteger> range = Range.openClosed(new
BigInteger(replicaInfo.start()),
+ new
BigInteger(replicaInfo.end()));
+ Map<String, List<String>> witnessesByDatacenter =
witnessesByDatacenter(replicationFactor, replicaInfo);
+ List<CassandraInstance> replicas = new ArrayList<>();
+ List<CassandraInstance> witnesses = new ArrayList<>();
+ for (Map.Entry<String, List<String>> byDatacenter :
replicaInfo.replicasByDatacenter().entrySet())
+ {
+ if (datacenter != null &&
!datacenter.equalsIgnoreCase(byDatacenter.getKey()))
+ {
+ continue;
+ }
+ List<String> datacenterWitnesses =
witnessesByDatacenter.getOrDefault(byDatacenter.getKey(),
+
Collections.emptyList());
+ 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 (datacenterWitnesses.contains(replicaKey))
+ {
+ witnesses.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 (!witnesses.isEmpty())
+ {
+ rangeWitnesses.put(range,
witnesses.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={} numRangesWithWitnesses={} "
+ + "numInstances={}", keyspace, rangeReplicas.size(),
rangeWitnesses.size(), instances.size());
+ return new CassandraRing(partitioner, keyspace, replicationFactor,
instances, rangeReplicas, rangeWitnesses);
+ }
+
+ /**
+ * Returns the witness replicas the Sidecar reported for a range, after
checking them against the replication
+ * factor.
+ * <p>
+ * A keyspace with transient replicas must have its witnesses reported: a
witness does not hold complete data, so
+ * without knowing which replica is the witness the reader could select it
and silently return incomplete
+ * results. Failing is the only safe option.
+ *
+ * @param replicationFactor the keyspace replication factor
+ * @param replicaInfo the replicas of one token range
+ * @return datacenter to witness replicas, empty when the range has none
+ * @throws IllegalStateException when witnesses are expected but not
reported, or do not match the replication
+ * factor
+ */
+ @VisibleForTesting
+ static Map<String, List<String>> witnessesByDatacenter(ReplicationFactor
replicationFactor,
+
TokenRangeReplicasResponse.ReplicaInfo replicaInfo)
+ {
+ Map<String, List<String>> witnessesByDatacenter =
replicaInfo.witnessesByDatacenter();
+ if (witnessesByDatacenter == null)
+ {
+ if (replicationFactor.hasTransientReplicas())
Review Comment:
Thanks Arjun! Added the test `WitnessReplicaBulkReaderTest` and fixed the
above case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
