Adding nodes to a stateless web server fleet is as simple as bumping an Auto-Scaling Group integer. Adding nodes to a stateful, globally distributed database while it continues serving thousands of queries per second requires precision orchestration.

Situation

In 2023, the 27-node Apache Cassandra 4.1 cluster (running on OpenJDK 11) reached 70% disk capacity across its 28 TB dataset. During traffic peaks, p99 read latency was doubling. We needed to expand the cluster horizontally to 36 nodes (moving from 9 to 12 nodes per Availability Zone across 3 AZs) without incurring downtime or dropping client mutations.

Unlike a stateless application tier, we could not simply attach 9 new nodes simultaneously. When a new Cassandra node joins the ring, existing nodes must stream gigabytes of SSTable data to the new arrival to satisfy its token assignments. Un-throttled token streaming would saturate instance network interfaces, starve client read traffic, and exhaust disk IOPS on the donor nodes.

The Problem

Cassandra distributes data using a consistent hashing ring. In modern Cassandra (4.x+), clusters use virtual nodes (vnodes) configured via num_tokens: 16. When a new node bootstraps, it randomly selects 16 token ranges and calculates which existing nodes own the data for those ranges. It then requests a copy of that data.

If we provisioned and bootstrapped all 9 new EC2 instances simultaneously, dozens of existing nodes would concurrently open heavy sequential read streams on their disks. This spikes PendingCompactions on donor nodes, pushes EBS volume utilization past 75%, and triggers streaming timeouts.

How do you safely extract 10 TB of data from an active 27-node cluster to 9 new nodes without degrading the primary read/write path?

Staged Zero-Copy Expansion

To scale out safely, the pattern architects a staged batch pipeline. We expanded the cluster in waves of 3 nodes (one node per AZ per batch): 27 → 30 → 33 → 36.

flowchart TD
    Start[Current State: 27 Nodes] --> Batch1[Batch 1: Add 3 Nodes]
    
    subgraph "Batch Execution (1 Node per AZ)"
        AZA[Bootstrap Node 1 in us-east-1a]
        AZB[Bootstrap Node 2 in us-east-1b]
        AZC[Bootstrap Node 3 in us-east-1c]
        
        AZA --> Stream[Zero-Copy Streaming]
        AZB --> Stream
        AZC --> Stream
        
        Stream --> Join[Nodes Join Ring]
    end

    Batch1 --> Gate1{Stabilization Gate: — Pending Tasks < 100 — No Dropped Mutations}
    Gate1 -->|Pass| Batch2[Batch 2: Add 3 Nodes]
    Batch2 --> Gate2{Stabilization Gate}
    Gate2 -->|Pass| Batch3[Batch 3: Add 3 Nodes]
    Batch3 --> Cleanup[Sequential `nodetool cleanup`]

We leveraged Cassandra 4.1’s Netty-based Zero-Copy Streaming. When an entire SSTable’s data falls within the requested token range, the donor node doesn’t need to deserialize it into Java objects — it uses Linux sendfile() to stream the raw bytes directly from the page cache to the network socket, drastically reducing JVM heap pressure and CPU overhead. With 16 vnodes per node, many SSTables only partially overlap any single requested range and fall back to the slower per-partition streaming path, so the realized speedup depends on how cleanly SSTable boundaries align with the ranges being streamed.

In Practice

The documented pattern for safe token range movement requires explicit network throttling. In cassandra.yaml, we clamped the outbound streaming throughput to prevent donor nodes from saturating their ENIs (Elastic Network Interfaces) or competing with inter-node SSL encryption overhead for client traffic:

# Limit outbound streaming to 200 Megabits/s
stream_throughput_outbound_megabits_per_sec: 200

During the bootstrap of a batch, we actively monitored progress using nodetool netstats:

$ nodetool netstats
Mode: BOOTSTRAPPING
Not sending any streams.
Receiving 6 files, 15488102400 bytes total. Already received 1 files, 3221225472 bytes total

Only after the 3 nodes in a batch had fully joined the ring, and nodetool tpstats showed PendingTasks on MutationStage and ReadStage had returned to baseline, did we proceed to the next batch.

Once the ring reached 36 nodes, the original 27 nodes still held data they were no longer responsible for — roughly 25% of their prior load, corresponding to the 9 new nodes added to the 36-node ring. To reclaim that disk space, the playbook executes nodetool cleanup. Crucially, the playbook runs this sequentially (one node per AZ at a time) during off-peak hours, as cleanup forces a massive sequential disk read-and-rewrite operation that can easily cause disk I/O exhaustion if run concurrently across the cluster.

Where It Breaks

Design ChoiceTradeoffMitigation
Batching SizeBootstrapping 1 node per AZ is safe but takes longer.Automate the gating process via a Python JMX polling script (scale_cluster_batch.py) to reduce human wait time.
Stream Throttling200 Mbps protects client latency but prolongs the bootstrap window.Adjust dynamically based on instance network baseline; 10Gbps instances can handle higher limits.
Failed BootstrapA bootstrapping node crashes midway, leaving dead tokens in gossip.Do not restart the broken node. Run nodetool removenode on its Host ID from a healthy node, wipe the broken node’s EBS volume, and reprovision fresh.

Modern Cassandra Context: Cassandra 5.0 Enhancements

Cassandra 5.0 improves the scale-out experience in a few concrete ways. The Unified Compaction Strategy (UCS) adapts its sharding to reduce compaction amplification after massive token range shifts. The introduction of Trie SSTables (BTI format) reduces the memory overhead required during streaming, and TLS 1.3 hardware acceleration on modern AWS Graviton3/4 instances substantially lowers — though doesn’t eliminate — the cryptographic CPU penalty previously associated with encrypted zero-copy transfers.

What to Do Next

  • Problem: Adding nodes to Cassandra forces massive data streaming that can overwhelm disks, CPUs, and networks, causing client timeouts.
  • Solution: Throttle outbound streaming, use 4.x zero-copy streaming, and batch expansions logically by fault domain.
  • Proof: Staged 3-by-3 expansion prevents donor-node saturation, ensuring p99 read latencies remain stable while gigabytes of data migrate across the network.
  • Action: Before expanding, set stream_throughput_outbound_megabits_per_sec and script your nodetool netstats validation gates. Never run nodetool cleanup on all nodes at once.