2. The ring

Every distributed database must answer one question before anything else: given a piece of data, which machines hold it? Cassandra’s answer is the thing everything else is built on.

The naive answer, and why it fails

The obvious approach is modulo hashing. With five nodes, put row k on node hash(k) mod 5.

It works perfectly until you add a sixth node. Now every key maps to hash(k) mod 6, and roughly five-sixths of all data must move. For a cluster holding 50 TB, that is a multi-day event that saturates your network and probably takes the cluster down with it.

Modulo hashing means growth is a crisis. Cassandra needs growth to be routine.

Consistent hashing

The fix, from the Dynamo paper, is to hash keys and nodes into the same space, and arrange that space in a circle.

The space is the range of a 64-bit signed integer: −2⁶³ to 2⁶³−1. A value in it is called a token. Bend that line into a ring so the maximum wraps around to the minimum.

Each node is assigned one or more tokens, which places it at points on the ring. To find where a row lives:

1. Hash its partition key to get a token.
2. Walk clockwise from that token until you hit a node.
3. That node owns the data.

                            token 0
                               │
                    ┌──────────┴──────────┐
                    │                     │
              ┌─────┴─────┐         ┌─────┴─────┐
              │  node A   │         │  node B   │
              │ token 0   │         │ token 2^61│
              └───────────┘         └───────────┘
                    │                     │
        −2^63 ──────┤                     ├────── row hashes to here
       (wraps to    │                     │       walks clockwise ──▶
        2^63−1)     │                     │       lands on node C
              ┌─────┴─────┐         ┌─────┴─────┐
              │  node D   │         │  node C   │
              │token −2^61│         │ token 2^62│
              └───────────┘         └───────────┘

Now add a node. It takes a position on the ring and claims the range between itself and its predecessor. Only that one range moves. Its neighbour hands over a slice; nobody else is affected. Adding the sixth node to a five-node cluster moves about one-sixth of the data, and that data comes from one place, streams in the background, and does not interrupt service.

Removing a node is the mirror image: its range is absorbed by its successor.

The partitioner

The hash function is the partitioner. Since Cassandra 1.2 the default is Murmur3Partitioner, and you should never change it — it is fast, distributes well, and changing it requires rebuilding the cluster from scratch.

SELECT token(user_id), user_id FROM users LIMIT 5;

That token() function is not a curiosity; it is occasionally the only way to page through an entire table (chapter 6).

The important property is that the hash is uniform and opaque. Keys that look adjacent to you — user_1000 and user_1001 — land at unrelated points on the ring. You cannot influence placement by naming things cleverly, and you should not try.

Virtual nodes

One token per node has two problems.

Uneven distribution. With eight randomly-placed tokens, some ranges are much bigger than others. One node ends up owning three times its share.

Slow, lopsided recovery. If each node owns one contiguous range, then when a node dies, its data is rebuilt from exactly the handful of nodes that replicate that range. Those few nodes do all the work while everyone else is idle.

Virtual nodes (“vnodes”) fix both. Each physical node is assigned many tokens — scattered around the ring — so it owns many small ranges instead of one big one.

   Without vnodes: node A owns one arc
   ┌───────────────────────────────────────────────┐
   │AAAAAAAA BBBBBBBB CCCCCCCC DDDDDDDD            │
   └───────────────────────────────────────────────┘
   A dies → only B and D (its neighbours) can rebuild it

   With vnodes: node A owns many scattered slivers
   ┌───────────────────────────────────────────────┐
   │ABDCADBCDABCADBCBADCDABCADBCABDC...            │
   └───────────────────────────────────────────────┘
   A dies → EVERY other node holds some of A's replicas
            and they all stream in parallel

Rebuilding a dead node goes from “two machines work for six hours” to “all twenty machines work for twenty minutes.” That difference is the main reason vnodes exist.

cassandra.yaml

num_tokens: 16
allocate_tokens_for_local_replication_factor: 3

The default was 256 for many years. Cassandra 4.0 changed the shipped default to 16, because 256 turned out to have real costs: repair works range by range, and 256 tokens × many nodes means a very large number of tiny ranges to coordinate, which made repair slow and the failure surface larger. Sixteen, combined with the token allocation algorithm, gives nearly as good a distribution with far less overhead.

allocate_tokens_for_local_replication_factor turns on the smart allocator — instead of picking tokens at random, it picks tokens that actively even out ownership. Set it. Random token selection with only 16 tokens per node produces noticeably lopsided clusters.

nodetool ring        # every token and its owner (long output)
nodetool status      # per node: state, load, and % of the ring owned

nodetool status is the one you will actually run. The Owns column should be close to even. If one node owns 40% of the ring in a five-node cluster, it will run out of disk first and serve more traffic than it should.

Replication

So far each range has one owner, which would mean losing a node loses data. Replication factor (RF) is how many copies exist.

The rule is simple: walk clockwise from the token and take the next RF distinct nodes. The first is sometimes called the “primary” replica, but the word is misleading — all replicas are equal. There is no leader, no special copy, and no ordering among them. The first one is just the first one you meet walking clockwise.

   RF = 3, row hashes into node B's range:

        A ──▶ B ──▶ C ──▶ D ──▶ E ──▶ (wraps to A)
              ▲     ▲     ▲
              └─────┴─────┴── the three replicas

RF is set per keyspace, not per cluster — so you can have one keyspace at RF=3 for important data and another at RF=1 for something disposable.

The two replication strategies

-- Never use this in production
CREATE KEYSPACE demo WITH replication = {
  'class': 'SimpleStrategy', 'replication_factor': 3
};

-- Always use this
CREATE KEYSPACE app WITH replication = {
  'class': 'NetworkTopologyStrategy',
  'dc_east': 3,
  'dc_west': 3
};

SimpleStrategy walks the ring and takes the next RF nodes, knowing nothing about racks or data centres. It is for single-node development and nothing else. It will cheerfully put all three of your replicas in the same rack.

NetworkTopologyStrategy (NTS) is topology-aware. It places replicas per data centre, and within a data centre it walks the ring skipping nodes whose rack it has already used. So with RF=3 in a data centre with three racks, you get one replica per rack — and losing an entire rack loses at most one copy of anything.

The syntax 'dc_east': 3, 'dc_west': 3 means three replicas in each, so six copies total. This is how multi-region works: each data centre has a full, locally-quorum-able set of replicas (chapter 12).

Use NetworkTopologyStrategy even with one data centre

Migrating from SimpleStrategy to NTS later means changing replica placement, which means streaming data and running repairs on a live cluster. It is entirely avoidable. Start with:

CREATE KEYSPACE app WITH replication = {
  'class': 'NetworkTopologyStrategy', 'dc1': 3
};

even when dc1 is your only data centre.

Racks

A “rack” is whatever you tell Cassandra it is: a physical rack, an AWS availability zone, a failure domain of your choosing. NTS treats it as “things that fail together” and spreads replicas across them.

The rule that matters: either use one rack for everything, or use exactly as many racks as your replication factor. Three racks with RF=3 gives a perfect one-replica-per-rack layout. Two racks with RF=3 gives an uneven layout where one rack holds two replicas of everything, so losing it costs you quorum. Five racks with RF=3 works but distributes unevenly during topology changes.

Where does this metadata live?

The token assignments, the schema, and the topology are all stored in local system keyspace tables on every node and propagated by gossip. There is no configuration server, no ZooKeeper, no etcd.

SELECT peer, data_center, rack, tokens FROM system.peers;
SELECT * FROM system.local;

Gossip

With no leader and no coordination service, how does a node learn that another node exists, has died, or has moved?

Gossip. Once per second, every node picks up to three other nodes and exchanges state with them: who is up, who is down, what tokens each holds, what version each runs, what its schema version is.

   every second, node A:
     ├─ gossips with a random live node
     ├─ maybe gossips with a random unreachable node (to detect recovery)
     └─ maybe gossips with a seed node

   each exchange is 3 messages:
     SYN  "here is my digest of what I know"
     ACK  "here is what you're missing, and my digest"
     ACK2 "here is what you're missing"

Information spreads exponentially. In a cluster of hundreds of nodes, a state change reaches everyone in a few seconds. There is no single point of failure because there is no single point of anything.

Seeds

A brand-new node has to talk to someone first. Seed nodes are the bootstrap contact points, listed in cassandra.yaml:

seed_provider:
  - class_name: org.apache.cassandra.locator.SimpleSeedProvider
    parameters:
      - seeds: "10.0.1.10,10.0.1.11,10.0.2.10"

Three things people get wrong about seeds, all of them worth stating plainly:

• Seeds are not special. They are not masters. They hold no extra data and have no extra authority. They are just addresses that new nodes know to call.
• List two or three per data centre, not all your nodes. A long seed list makes gossip noisier without helping.
• A node listed as its own seed will not bootstrap. It assumes it is the first node in a new cluster and skips streaming data. This is a genuinely common mistake when adding capacity — the new node joins the ring, owns a range, and serves empty results for it. Never put a joining node in its own seed list.

Failure detection

Cassandra does not use a fixed timeout to decide a node is dead. It uses the Phi Accrual Failure Detector, which tracks the recent history of gossip arrival times from each peer and outputs a continuously-rising suspicion level rather than a boolean.

The advantage is adaptivity. A node on a reliable local network is declared down quickly; a node across a flaky WAN link gets more slack, because the detector has learned that its heartbeats are irregular.

phi_convict_threshold: 8    # default. Raise to 10-12 on unreliable networks.

Marking a node down is a local decision each node makes independently. There is no vote, no quorum, and no consensus about liveness — which is exactly why a partition does not stop anything.

Coordinators

Any node can receive any request. The node the client connects to becomes the coordinator for that request.

   client ──▶ node E  (coordinator)
                 │
                 │ E is not a replica for this key.
                 │ E computes the token, finds replicas B, C, D,
                 │ and forwards the request to them.
                 ▼
              B  C  D
                 │
                 │ replies come back to E
                 ▼
   client ◀── node E

The coordinator’s job is: hash the key, find the replicas, send the request, wait for as many responses as the consistency level requires, resolve any disagreement, reply.

This is why there is no “connect to the primary” step in Cassandra and no connection-routing infrastructure. Every node can serve every request.

Token-aware routing

If the coordinator happens to be a replica, one network hop disappears. Good drivers exploit this: they maintain their own copy of the ring, hash the partition key client-side, and send the request straight to a replica.

from cassandra.policies import TokenAwarePolicy, DCAwareRoundRobinPolicy
cluster = Cluster(
    contact_points=['10.0.1.10'],
    load_balancing_policy=TokenAwarePolicy(
        DCAwareRoundRobinPolicy(local_dc='dc_east')
    ),
)

This is close to mandatory, and it only works if you use prepared statements — the driver needs the parsed statement to know which bound parameter is the partition key. A driver using unprepared string queries cannot do token-aware routing and will bounce every request through an extra hop.

DCAwareRoundRobinPolicy with an explicit local_dc is the other half: it stops your application in Virginia from load-balancing across to Sydney.

Seeing the ring

# The overview you will run most often
nodetool status
# UN = Up Normal, DN = Down Normal, UJ = Up Joining, UL = Up Leaving
#
# Datacenter: dc_east
# --  Address     Load     Tokens  Owns    Host ID   Rack
# UN  10.0.1.10   142 GiB  16      33.4%   abc-123   rack1
# UN  10.0.1.11   139 GiB  16      33.2%   def-456   rack2
# UN  10.0.1.12   145 GiB  16      33.4%   ghi-789   rack3

# Which nodes hold this specific key? Extremely useful for debugging.
nodetool getendpoints my_keyspace my_table 'some-partition-key'

# What does this node think of everyone else?
nodetool gossipinfo

# Token ranges and their replicas
nodetool describering my_keyspace

What to take from this chapter

• Consistent hashing puts keys and nodes in the same circular token space. Adding a node moves only one node’s worth of data.
• Murmur3Partitioner is the default and you will never change it. Placement is uniform and opaque — you cannot steer it.
• Vnodes (default 16 since 4.0) give even distribution and let a dead node be rebuilt by the whole cluster in parallel. Turn on allocate_tokens_for_local_replication_factor.
• Replication walks clockwise for RF distinct nodes. All replicas are equal; there is no primary.
• Always NetworkTopologyStrategy, even with one data centre. Use one rack, or exactly RF racks.
• Gossip spreads state in seconds with no coordination service. Seeds are bootstrap addresses, not masters — and never seed a node to itself.
• Any node can coordinate any request. Use token-aware routing plus prepared statements to skip a hop.

Next: what those nodes actually write to disk.