Skip to content

AxonOps — AI-Native Control Plane for Open Source Data Platforms

Cassandra Distributed Data

Cassandra distributes data across multiple nodes to achieve fault tolerance and horizontal scalability. This section covers the three interconnected mechanisms that govern how data is distributed, replicated, and accessed consistently across the cluster.


Cassandra's distributed architecture derives from Amazon's Dynamo paper (DeCandia et al., 2007, "Dynamo: Amazon's Highly Available Key-value Store"), which introduced techniques for building highly available distributed systems.

Dynamo ConceptPurposeCassandra Implementation
Consistent hashingDistribute data across nodesToken ring with partitioners
Virtual nodesEven distribution, incremental scalingvnodes (num_tokens)
ReplicationFault toleranceConfigurable replication factor
Sloppy quorumAvailability during failuresHinted handoff
Vector clocksConflict resolutionTimestamps (last-write-wins)
Merkle treesEfficient synchronizationMerkle tree synchronization
Gossip protocolFailure detection, membershipGossip protocol
CAP theoremConsistency/availability trade-offTunable consistency per operation

Cassandra implements most Dynamo concepts but makes different trade-offs in some areas.

Timestamps vs Vector Clocks

Cassandra uses timestamps for conflict resolution rather than vector clocks, choosing simplicity over preserving all conflicting versions. This means concurrent writes to the same key result in last-write-wins semantics based on timestamp, rather than preserving both versions for application-level resolution.


Cassandra's defining architectural characteristic is its masterless (or "peer-to-peer") design. Every node in a Cassandra cluster is identical in function—there is no primary, leader, master, or coordinator node that holds special responsibility for the cluster. Unlike systems that require distinct node roles (primary/replica, leader/follower, master/slave), Cassandra nodes are homogeneous: same configuration, same binary, same capabilities. This architectural simplicity translates directly to operational simplicity—adding capacity means deploying identical nodes, and any node can be replaced without special failover procedures.

Traditional Master-Based System:

MasterBased • Writes MUST go to primary • Primary failure requires election/failover • Replicas are read-only Client Client Primary PRIMARY (Master) Client->Primary writes Replica1 REPLICA 1 (read-only) Primary->Replica1 replicates Replica2 REPLICA 2 (read-only) Primary->Replica2 replicates Replica1->Client reads Replica2->Client reads

Cassandra Masterless Design:

Masterless • ANY node accepts reads AND writes • No election required on failure • All nodes are equal peers A Node A B Node B A--B C Node C A--C B--C D Node D B--D C--D E Node E C--E F Node F C--F D--E D--F E--A E--F F--A F--B Client Client Client--A r/w Client--C r/w Client--E r/w

In Cassandra:

  • Any node can serve as coordinator for any request
  • The coordinator role is assigned per-request, not per-cluster
  • In traditional Cassandra, no node holds special metadata or routing responsibility (note: CEP-21 introduces cluster metadata services in newer versions)
  • Cluster continues operating if any node (or multiple nodes) fails
SystemArchitectureWrite PathFailure Behavior
CassandraMasterlessAny node accepts writes for any partitionNo failover needed; remaining nodes continue
MongoDBPrimary/SecondaryWrites to primary only; primary replicates to secondariesElection required; brief write unavailability
MySQL (Group Replication)Single-primary or Multi-primarySingle-primary: one node; Multi-primary: any nodePrimary election on failure
PostgreSQL (Streaming)Primary/StandbyPrimary only; standbys are read-onlyManual or automatic failover required
CockroachDBRaft consensusLeader per range; leader accepts writesRaft leader election per range
TiDBRaft consensusLeader per region; leader accepts writesRaft leader election per region
Redis ClusterPrimary/Replica per slotPrimary for each hash slotFailover election per slot

Cassandra drivers establish connections to all nodes in the local datacenter (and optionally remote DCs). The driver maintains a connection pool to each node and load balances requests across them. For each request:

1. Driver selects a coordinator node (load balancing across all connected nodes)
2. That node becomes the COORDINATOR for this request
3. Coordinator determines which nodes hold the data (using token ring)
4. Coordinator forwards request to appropriate replica nodes
5. Coordinator collects responses and returns result to client
Request 1: Client → Node A (coordinator) → Nodes B, C, D (replicas)
Request 2: Client → Node C (coordinator) → Nodes A, E, F (replicas)
Request 3: Client → Node B (coordinator) → Nodes C, D, A (replicas)
Each request can use a different coordinator.
No node is "special" or required for cluster operation.
CoordinatorFlow cluster_replicas Replica Nodes for partition Client Client Coordinator Node B (Coordinator for this request) Client->Coordinator 1. Request Coordinator->Client 4. Response (after quorum) R1 Node A (Replica 1) Coordinator->R1 2. Forward R2 Node D (Replica 2) Coordinator->R2 2. Forward R3 Node F (Replica 3) Coordinator->R3 2. Forward R1->Coordinator 3. ACK R2->Coordinator 3. ACK
BenefitDescription
No single point of failureAny node can fail without affecting cluster availability
No failover delayNo election process; operations continue immediately
Write scalabilityAll nodes accept writes; adding nodes generally increases write capacity
Simpler operationsNo primary/replica distinction to manage
Geographic distributionEach datacenter is autonomous; no cross-DC leader election
Trade-offDescriptionMitigation
Conflict resolutionConcurrent writes to same key can conflictLast-write-wins (timestamps); LWT for critical operations
No single source of truthNo authoritative primary for readsQuorum reads; repair for convergence
Coordination overheadEach request requires multi-node coordinationToken-aware routing; LOCAL_* consistency levels
Complexity in orderingNo global write orderingPer-partition ordering; application-level sequencing

Consistency Guarantees: Master-Based vs Masterless

Section titled “Consistency Guarantees: Master-Based vs Masterless”

Master-based architectures achieve consistency through a single authoritative node—but at the cost of availability during failures and a write throughput ceiling. Cassandra provides the same guarantees when needed, while allowing flexibility to optimize for availability or performance when strong consistency is not required.

RequirementMaster-Based ApproachCassandra Approach
Strong consistencyAll writes through primary (single point of failure, write bottleneck)SERIAL consistency via Paxos (no single point of failure, scales horizontally)
Conflict resolutionPrimary is authoritative (unavailable during failover)Last-write-wins by default; LWT for compare-and-set when needed
Read-your-writesRead from primary (adds latency, primary overload risk)QUORUM reads + writes (R + W > N), load balanced across replicas

Key Difference

Master-based systems force strong consistency at all times with corresponding availability trade-offs. Cassandra allows per-query tuning—strong consistency for financial transactions, eventual consistency for analytics or caching.

Cassandra's masterless design extends across datacenters:

  • Single cluster spans multiple datacenters with topology-aware replication
  • No cross-DC leader election required
  • LOCAL_QUORUM enables DC-local consistency without cross-DC coordination
  • Cluster continues operating if one DC fails entirely
MultiDCMasterless cluster_dc1 data center alpha RF = 3 cluster_rack1_dc1 rack 1 cluster_rack2_dc1 rack 2 cluster_rack3_dc1 rack 3 cluster_dc2 data center omega RF = 3 cluster_rack1_dc2 rack 1 cluster_rack2_dc2 rack 2 cluster_rack3_dc2 rack 3 N1_1 node 1 N1_6 node 6 N1_2 node 2 N1_5 node 5 N1_3 node 3 N1_4 node 4 N2_4 node 4 N1_4:e->N2_4:e multi-DC replication N2_1 node 1 N2_6 node 6 N2_2 node 2 N2_5 node 5 N2_3 node 3

Multi-DC Latency Advantage

Systems using Raft or Paxos consensus across datacenters require cross-DC communication for every write, adding latency proportional to the distance between datacenters. Cassandra with LOCAL_QUORUM avoids this cross-DC round-trip for most operations.


Partitioning determines which node stores a given piece of data. Cassandra uses consistent hashing to map partition keys to tokens, and tokens to nodes on a ring.

Partition Key → Hash Function → Token → Node(s)
Example:
"user:123" → Murmur3Hash → -7509452495886106294 → Node B

The partitioner (hash function) ensures even data distribution regardless of key patterns. This prevents hot spots where sequential keys would otherwise concentrate on a single node.

See Partitioning for details on consistent hashing, the token ring, and partitioner options.

Replication copies each partition to multiple nodes for fault tolerance. The replication factor (RF) determines how many copies exist.

ReplicationRF3 NodeA Node A (Copy 1) NodeB Node B (Copy 2) NodeC Node C (Copy 3) Partition Partition (user:123) Partition->NodeA Partition->NodeB Partition->NodeC

The replication strategy determines how replicas are placed—whether they respect datacenter and rack boundaries to survive infrastructure failures.

See Replication for details on strategies, snitches, and configuration.

Consistency determines how many replicas must acknowledge reads and writes. Because replicas may temporarily diverge, the consistency level controls the trade-off between consistency, availability, and latency.

Write with QUORUM (RF=3):
- Send write to all 3 replicas
- Wait for 2 acknowledgments (majority)
- Return success to client
Read with QUORUM (RF=3):
- Contact 2 replicas
- Compare responses, return newest
- Propagate newest version to stale replicas

Strong Consistency Formula

The formula R + W > N (reads + writes > replication factor) guarantees that reads see the latest writes. With RF=3, using QUORUM (2) for both reads and writes satisfies this: 2 + 2 = 4 > 3.

See Consistency for details on consistency levels and guarantees.

When replicas diverge due to failures or timing differences, synchronization mechanisms detect the divergence and propagate missing data to restore convergence:

MechanismTriggerFunction
Hinted handoffWrite to unavailable replicaDeferred delivery when replica recovers
Read reconciliationQuery executionPropagates newest version to stale replicas
Merkle tree synchronizationScheduled maintenanceFull dataset comparison and convergence

See Replica Synchronization for details on convergence mechanisms.


A single write operation involves all three mechanisms:

INSERT INTO users (id, name) VALUES (123, 'Alice')
WITH CONSISTENCY QUORUM
1. PARTITIONING
Coordinator hashes partition key:
token(123) = -7509452495886106294
2. REPLICATION
Look up replicas for this token:
RF=3, NetworkTopologyStrategy → Nodes A, B, C (different racks)
3. CONSISTENCY
Send write to all replicas, wait for QUORUM (2):
Node A: ACK ✓
Node B: ACK ✓
Node C: (still writing, but QUORUM met)
Return SUCCESS to client
4. ANTI-ENTROPY
If Node C was temporarily down:
- Coordinator stores hint
- When C recovers, hint is delivered
- If hints expire, scheduled synchronization restores convergence

The CAP theorem, formulated by Eric Brewer in 2000 and proven by Gilbert and Lynch in 2002 (Gilbert & Lynch, 2002, "Brewer's Conjecture and the Feasibility of Consistent, Available, Partition-Tolerant Web Services"), states that a distributed data store can provide at most two of three guarantees simultaneously:

CAP C Consistency Every read receives the most recent write or an error A Availability Every request receives a response (no errors) P Partition Tolerance System continues operating despite network failures CAP CAP Theorem: Choose 2 of 3 CAP->C CAP->A CAP->P
PropertyDefinitionImplication
Consistency (C)All nodes see the same data at the same timeReads always return the most recent write
Availability (A)Every request receives a non-error responseNo request times out or returns failure
Partition Tolerance (P)System continues operating despite network partitionsNodes can be split into groups that cannot communicate

Network Partitions Are Inevitable

In real distributed systems, network partitions are inevitable—switches fail, cables are cut, datacenters lose connectivity. A system that cannot tolerate partitions is not a distributed system; it is a single-node system with remote storage.

Therefore, the practical choice is between:

  • CP (Consistency + Partition Tolerance): During a partition, reject operations that cannot guarantee consistency
  • AP (Availability + Partition Tolerance): During a partition, continue accepting operations even if nodes may diverge
CAPChoice partition Network Partition Occurs cp CP Choice Reject writes to maintain consistency → Some requests fail partition->cp prioritize consistency ap AP Choice Accept writes on both sides of partition → Temporary inconsistency partition->ap prioritize availability
CategoryBehavior During PartitionExamples
CPReject operations that cannot be consistently appliedPostgreSQL, MySQL, MongoDB (default), CockroachDB, Spanner
APAccept operations; resolve conflicts laterCassandra (default), DynamoDB, Riak, CouchDB

Cassandra is typically classified as AP—it prioritizes availability over consistency by default. However, this classification oversimplifies Cassandra's capabilities.

Tunable, Not Fixed

Unlike most databases that are permanently CP or AP, Cassandra allows choosing the consistency-availability trade-off on a per-operation basis. The same cluster can serve AP workloads (analytics, caching) and CP workloads (transactions, user data) simultaneously.

Default behavior (AP):

  • Writes succeed if any replica is available
  • Reads return data even if replicas disagree
  • Conflicts resolved by timestamp (last-write-wins)

With tunable consistency, Cassandra can behave as CP:

  • QUORUM reads and writes ensure R + W > N (overlapping quorums)
  • ALL requires all replicas to respond
  • SERIAL provides linearizable consistency via Paxos

Tunable Consistency: Per-Operation CAP Position

Section titled “Tunable Consistency: Per-Operation CAP Position”

Unlike databases that enforce a single consistency model, Cassandra allows choosing the consistency-availability trade-off for each operation:

Consistency LevelCAP PositionBehavior During Partition
ANYAPWrite succeeds if any node (including coordinator) receives it
ONEAPWrite/read succeeds if one replica responds
QUORUMCPRequires majority of replicas; may reject if quorum unavailable
ALLCPRequires all replicas; rejects if any replica unavailable
SERIALCPLinearizable via Paxos; rejects if consensus cannot be reached

Practical implications:

Partition splits cluster: Nodes {A, B} | {C, D, E}
RF = 3, replicas on nodes A, C, E
With QUORUM (requires 2 of 3 replicas):
- Left side (A): Can reach 1 replica → QUORUM fails
- Right side (C, E): Can reach 2 replicas → QUORUM succeeds
- Result: CP behavior, partial availability
With ONE:
- Both sides can reach at least 1 replica
- Result: AP behavior, full availability, possible inconsistency

CAP Only Applies During Partitions

The CAP theorem applies specifically during network partitions. During normal operation (no partitions), Cassandra provides both consistency and availability—the trade-off only manifests when partitions occur.

StateConsistencyAvailabilityNotes
Normal operation✓ (with QUORUM)No trade-off required
During partitionChoose oneChoose oneCAP trade-off applies
After partition heals✓ (eventually)Repair restores consistency

The PACELC theorem (Abadi, 2012) extends CAP to address behavior during normal operation:

Partition → Availability vs Consistency Else → Latency vs Consistency

During normal operation, the trade-off is between latency and consistency:

SystemDuring Partition (PAC)Normal Operation (ELC)
Cassandra (ONE)PAEL (low latency, eventual consistency)
Cassandra (QUORUM)PCEC (higher latency, strong consistency)
PostgreSQLPCEC
DynamoDBPAEL

Cassandra's tunable consistency allows choosing different PACELC positions for different operations within the same cluster.


SectionDescription
PartitioningConsistent hashing, token ring, partitioners
ReplicationStrategies, snitches, replication factor
ConsistencyConsistency levels, guarantees, LWT
Consensus and PaxosConsensus algorithms and Paxos for lightweight transactions
Replica SynchronizationHinted handoff, read reconciliation, Merkle trees
Secondary Index QueriesDistributed query execution with indexes
Materialized ViewsDistributed MV coordination and consistency challenges
Data StreamingBootstrap, decommission, repair, and hinted handoff streaming
CountersDistributed counting, CRDTs, counter operations