Phase 1: Distributed Architecture, Dynamo Ring & Gossip Protocol¶

Audience: Principal Distributed Systems Architect & Kernel Reliability Engineer
Source References: src/java/org/apache/cassandra/dht/, src/java/org/apache/cassandra/gms/

1. The Distributed Token Ring & Murmur3Partitioner¶

Cassandra organizes all physical cluster nodes into a logical circular ring using Consistent Hashing:

The Murmur3Partitioner 64-Bit Token Space:
┌────────────────────────────────────────────────────────────────────────┐
│ Range: -9,223,372,036,854,775,808 (-2^63) to +9,223,372,036,854,775,807 │
└────────────────────────────────────────────────────────────────────────┘

Distributed Consistent Hash Ring:
                 Token -2^63 / +2^63-1
                         │
        Node D ──────────┼────────── Node A
    (Tokens: 4B -> 9B)   │       (Tokens: -9B -> -4B)
                         │
                         │
        Node C ──────────┼────────── Node B
    (Tokens: 0 -> 4B)    │       (Tokens: -4B -> 0)

1.1 Key Placement Equation¶

When an application inserts a row with partition key user_id = "alice": $$\text{Token} = \text{MurmurHash3}(\text{"alice"})$$ - The coordinator locates the first node whose assigned token range covers that value. - Replicas are selected by walking clockwise around the ring until $N$ distinct physical nodes (controlled by Replication Factor) are assigned.

1.2 Virtual Nodes (Vnodes)¶

In early Cassandra versions, each physical machine owned a single monolithic token range on the ring. Replacing a dead node required copying contiguous gigabytes of data from its immediate neighbors, saturating specific network links.

With Virtual Nodes (Vnodes): - Each physical node is assigned $M$ small discrete tokens distributed uniformly across the entire ring (typically num_tokens: 128 or 16). - When a new physical node joins the cluster: - It acquires small token slices from every other node in the cluster. - Bootstrap data streaming occurs in parallel across all nodes simultaneously!

2. Gossip Protocol: Decentralized Cluster Discovery¶

Cassandra has no central master, coordinator registry, or ZooKeeper dependency. Cluster membership and node topology are maintained through a peer-to-peer Gossip Protocol (org.apache.cassandra.gms.Gossiper):

The 3-Way Scuttlebutt Gossip Handshake:
Every 1,000 milliseconds, Node A selects a random peer (Node B):

1. Node A ──► Sends GossipDigestSynMessage
              (Summary of highest generation and version numbers known to A)

2. Node B ──► Evaluates Digest:
              - Computes deltas where B is newer than A
              - Computes deltas where A is newer than B
              - Sends GossipDigestAckMessage (Contains B's newer states + requests A's deltas)

3. Node A ──► Updates its internal state with B's deltas
          ──► Sends GossipDigestAck2Message (Sends missing deltas to B)

2.1 Anatomy of Node State & Heartbeat¶

Each node maintains a HeartBeatState: - Generation Number: Unix timestamp recorded when the Cassandra daemon booted. Monotonically increasing; increments only on restart. - Version Number: Monotonically increasing atomic integer incremented on every internal state change (status, tokens, schema version, load).

3. The Phi Accrual Failure Detector¶

Traditional distributed systems detect dead nodes using binary heartbeat timeouts ("If no reply in 5 seconds, declare dead"). In heterogeneous cloud environments, intermittent network packet loss or transient JVM GC pauses trigger false failure declarations.

Cassandra implements the Phi Accrual Failure Detector (Hayashibara et al.):

Failure Detection Philosophy:
Instead of asking "Is the node dead?",
it asks: "What is the continuous mathematical probability that this node is dead?"

3.1 The Mathematical Model¶

For every node, Cassandra maintains a sliding window of the last 1,000 heartbeat inter-arrival intervals: $\{I_1, I_2, \dots, I_m\}$.
It fits these intervals to a normal distribution with mean $\mu$ and variance $\sigma^2$.
When time $t$ has elapsed since the last received heartbeat, it calculates: $$P_{\text{later}}(t) = \text{Probability that a heartbeat would arrive after time } t$$
Computes the dynamic scale metric $\Phi$ (Phi): $$\Phi = -\log_{10}(P_{\text{later}}(t))$$

Interpretation of Phi:
Phi = 1  ===> 1 in 10 chance of false alarm.
Phi = 2  ===> 1 in 100 chance of false alarm.
Phi = 8  ===> 1 in 100,000,000 chance of false alarm.

Production Configuration (`cassandra.yaml`):¶

phi_convict_threshold: 12 # Mark node DOWN when Phi crosses 12

(A threshold of 12 provides enterprise resilience against transient network jitter and 2-second GC pauses while decisively convicting dead hardware within 10 to 15 seconds).

4. Snitches & Topology Routing¶

The Snitch informs Cassandra of the physical network topology (which rack and datacenter each node belongs to).

NetworkTopologyStrategy Replication Across Two Datacenters:
Replication: { 'us-east-1': 3, 'eu-central-1': 3 }

Token Ring Navigation:
- Walks clockwise around the ring.
- First replica placed on natural node.
- Subsequent replicas are placed on nodes residing in DIFFERENT RACKS within the datacenter
  to survive complete physical top-of-rack switch failures!

Production Standard: `GossipingPropertyFileSnitch`¶

Nodes read their local rack and datacenter assignments from cassandra-rackdc.properties and automatically broadcast them to the entire cluster via Gossip.