CASSANALYTICS-197: Source token ranges from Cassandra for mutation tracked keyspaces - #244
Open
mansikhara wants to merge 10 commits into
Open
mansikhara wants to merge 10 commits into
mansikhara wants to merge 10 commits into
Conversation
added 10 commits
August 28, 2026 16:46
…r witness-enabled keyspaces Cassandra accepts a replication factor of the form <replicas>/<transient>, which witness replicas under mutation tracking reuse, so a witness-enabled keyspace declares 'datacenter1': '3/1' meaning three replicas of which one is a witness. CqlUtils.extractReplicationFactor() passed each datacenter value to Integer.parseInt, so a bulk read against any such keyspace failed during job setup with an uncaught NumberFormatException. Transient counts are now tracked per datacenter alongside the existing totals and exposed through getFullReplicationFactor(), getTransientReplicationFactor(), getFullReplicas(dc), getTransientReplicas(dc) and hasTransientReplicas(). getTotalReplicationFactor() keeps its existing meaning of all replicas including witnesses, so behaviour for untracked keyspaces is unchanged. Parsing applies the same constraints Cassandra enforces in locator.ReplicationFactor.validate. A new parseStrict factory reports an unparseable or empty replication map at parse time rather than dropping the datacenter and failing later with a misleading "DC not found in replication factor"; the lenient constructor is retained unchanged for CDC callers. The Kryo serializer and CassandraRing's hand-rolled JDK readObject/writeObject are updated so the new field survives serialization to Spark executors. CassandraRing gains a pinned serialVersionUID and a format-version byte, because the class signature did not change and an older stream would otherwise be silently misread rather than rejected. Prerequisite for CASSANALYTICS-164. patch by Mansi Khara; reviewed by TBD for CASSANALYTICS-194
CassandraRing derives token ranges locally from tokens and the replication factor. That derivation ignores racks, while Cassandra's replica assignment is rack aware, so the derived ranges are only guaranteed to match Cassandra for a single rack per datacenter. Witness replica support needs replica information per real token range, so the derived ranges cannot be relied on. Adds an optional range-to-replicas mapping. When supplied, init() uses it instead of deriving; when absent the behaviour is unchanged, so untracked reads and the CDC caller are unaffected. Both hand-rolled serializers carry the new field and the JDK format version is bumped to 2. This is groundwork only. Nothing supplies the mapping yet.
For a tracked keyspace a range may be replicated to a witness that holds no data, so the reader must know which instance replicates which range. The locally derived ranges cannot answer that: the derivation ignores racks while Cassandra's replica assignment is rack aware, so the two only agree for a single rack per datacenter. When the keyspace is tracked, the range to replica mapping is now taken from the token-range-replicas endpoint (readReplicas, so pending ranges are excluded). Untracked keyspaces keep using the ring response exactly as before. Node discovery still comes from the ring response in both cases, so snapshot creation, sizing and the Sidecar client pool are untouched. Only the range to replica mapping changes. Replicas are reported as "address:port" and instances are keyed by fqdn, so the join goes through the replica metadata. A replica absent from the ring is skipped with a warning, since the ring is the source of truth for which nodes the reader can reach. A node owning multiple tokens is rejected, because it cannot be represented as a single CassandraInstance; mutation tracking requires num_tokens=1 so this should not arise. Adds a forceCassandraTokenRanges option to exercise the new path against untracked keyspaces. Tracked keyspaces need Cassandra 6.0 bridge modules before they can be created in integration tests, so without this the path would merge with no integration coverage.
Covers createCassandraRingFromTokenRangeReplicas: the address-to-fqdn join through the replica metadata, datacenter filtering, skipping a replica absent from the ring, rejecting a node that owns multiple tokens, and failing when Cassandra reports no read replicas. The fixture uses a range boundary the local derivation cannot produce, so the tests fail if the Cassandra-reported ranges are ignored. Also documents that forceCassandraTokenRanges is set lowercased, since Spark lowercases option keys and MapUtils looks them up that way.
…ck-aware cluster BulkReaderCassandraTokenRangesTest exercises the new topology path against an ordinary keyspace using forcecassandratokenranges, since tracked keyspaces cannot be created until Cassandra 6.0 bridge modules land. It asserts the full dataset rather than a row count, because an incomplete read is the failure mode the path exists to prevent, and covers LOCAL_QUORUM and a pushdown filter. BulkReaderMultiRackTest closes a coverage gap. Every existing test places all nodes of a datacenter in one rack, where Cassandra's rack-aware assignment reduces to taking the next RF nodes in ring order, which is what the reader derives. The cluster here uses six nodes across three racks with two adjacent nodes per rack and RF 3, so ring order starts 1,2,3 while Cassandra must pick 1,3,5 to satisfy one replica per rack. The two topology sources therefore genuinely disagree. One node per rack, or round-robin racks, would coincide and prove nothing. CassandraCluster only applied dcAndRackSupplier when dcCount was greater than one, so a single-datacenter multi-rack topology was silently ignored and fell back to the default rack. An explicit supplier is now always honoured; the built-in two-datacenter default is unchanged.
Four nodes is the minimum that makes the two topology sources disagree, so the test no longer asks for six. NetworkTopologyStrategy computes acceptableRackRepeats as RF minus rackCount, which is zero with three racks and RF 3, so a node whose rack was already used is skipped: for the range starting at node 1 the reader derives ring order 1,2,3 while Cassandra picks 1,3,4. Six nodes added no coverage and needed two more loopback aliases.
A range whose replicas were all skipped, or a topology response that simply omits part of the ring, previously produced a ring with a gap and no error. CassandraRing seeds its range map across the whole ring, so the uncovered range is present but with an empty replica list, and the reader only discovers this once a Spark partition covering it fails with NotEnoughReplicasException - which reads as a cluster availability problem rather than an incomplete topology. Note that TokenPartitioner's coverage assertion does not catch it either. Also deduplicates the replicas recorded per range. PartitionedDataLayer counts replicas per range to decide whether the consistency level is satisfied, so a duplicate would inflate that count and could let a read proceed at a consistency level it has not actually met.
Four formatting violations in the new tests: whitespace after '{' in three array initialisers,
and a brace left on the same line as a method signature.
Also records what BulkReaderMultiRackTest does and does not establish. Every replica in it is a
full replica holding data, so both replica selections return all rows and the reads agree even
though Cassandra and the reader choose different replicas. It guards the range mapping and the
replica join against regression; it does not show that witness replicas are handled, because
choosing the wrong replicas only loses data once some of them hold none. A green run should not
be read as covering the witness case.
…izer index widths The class javadoc asserted that token ranges are calculated assuming racks are not in use. That is now only true of the local derivation, and it is exactly the assumption the Cassandra-supplied ranges exist to avoid, so it described the opposite of the new behaviour. It now distinguishes the two paths. The JDK serializer wrote replica counts and indexes as shorts while the Kryo serializer used ints. Not a defect, since the instance count is itself a short so indexes cannot exceed its range, but the two paths have to stay in step and the asymmetry would silently truncate if that width ever widened. Both now use int.
mansikhara
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
https://issues.apache.org/jira/browse/CASSANALYTICS-197
The bulk reader derives token ranges itself from node tokens plus the replication factor. CassandraRing documents that this assumes racks are not in use, but Cassandra's replica
assignment is rack aware (NetworkTopologyStrategy.acceptableRackRepeats), so the two only agree when a datacenter has a single rack.
For a mutation tracked keyspace a range may be replicated to a witness that holds no data, so the reader must know which instance replicates which range. The derived ranges cannot
answer that reliably.
Changes
version is bumped
availability problem
Testing
serialization round-trips
different replicas. Actual witness behaviour needs a tracked keyspace and waits on CASSANALYTICS-192