Messaging and Failure Handling

Verb is an enum that defines every message type the cluster exchanges. Each Verb entry carries metadata: its expected Stage, serializer, expiration timeout, and handler class. Examples of verbs:

When adding a new message type, a new Verb entry is required. This is an intentional registry: it makes the full surface area of internode messages auditable in one place.

Message<T> is the typed envelope for all messages. It carries the Verb, source endpoint, creation timestamp, payload, and optional tracing and parameter headers. IVerbHandler<T> is the interface each verb’s handler implements — it receives the deserialized Message<T> and acts on it.

Handler registration happens in MessagingService.registerHandlers(). Every Verb has exactly one IVerbHandler implementation. When implementing a new message type, you provide both the serializer for the payload and an IVerbHandler that processes it on the receiving end.

Inbound messages are dispatched to Stage-specific thread pools defined in the Stage enum (org.apache.cassandra.concurrent). Key stages:

Thread pool sizes are configurable and observable via JMX and nodetool tpstats. When adding a new verb, assign it to the most appropriate existing stage. Introducing a new stage requires justification because it creates a new JMX surface and a new tuning knob for operators.

When a node first needs to send a message to a peer, OutboundConnection opens a TCP connection and performs a handshake. The handshake negotiates:

Cassandra supports rolling upgrades, which means a cluster may temporarily contain nodes running different versions. MessagingService.current_version tracks the local node’s messaging version. When sending to a peer, the outbound connection uses the negotiated version for that peer. Every message payload serializer must be versioned accordingly via IVersionedSerializer<T>. Changes to serialized types must either use version-conditional logic or increment the messaging version. Skipping this step causes silent data corruption or deserialization errors during rolling upgrades.

Each frame on the wire consists of:

Compression (LZ4) wraps the frame when enabled, not individual payloads.

OutboundConnections manages three logical channels to each peer, kept as separate TCP connections:

Each outbound connection has a bounded queue. When the queue fills (because the peer is slow or unavailable), MessagingService drops non-urgent messages and increments the dropped_messages metric. Contributors adding high-volume messages must consider whether the message type can be dropped safely under load.

Each gossip round is a three-way exchange:

Each round, a node also probabilistically contacts an unreachable node and always contacts a seed node if no seed was contacted in the random selection.

EndpointState is the data structure each node advertises for itself and relays for its peers. It contains:

ApplicationState keys the state a node advertises about itself:

VersionedValue wraps each application state value with a logical version number. Gossip uses the version to discard stale updates: a value with a lower version than the currently known value is ignored.

Gossip is eventually consistent. State changes — including node join, token changes, and schema version updates — propagate on gossip timescales: typically a few seconds in small clusters, longer in large or geographically distributed clusters. Contributors must never assume gossip state is immediately visible to all nodes after a change.

In Cassandra 6, ClusterMetadata (TCM) supersedes gossip for schema and topology state distribution. Gossip continues to carry liveness state (STATUS, heartbeats, DC, RACK) but is no longer the authoritative source for schema version or token assignments.

The failure detector maintains a sliding window of inter-arrival times for heartbeat messages from each peer. It models the distribution of those inter-arrival times as a normal distribution and computes:

A higher φ indicates lower probability that the peer is alive. When φ exceeds phi_convict_threshold (default: 8, configurable in cassandra.yaml), the node is convicted as DOWN.

phi_convict_threshold controls the sensitivity of failure detection:

Operators on cloud environments with variable network latency often raise this value.

DOWN means the failure detector has convicted the node — it is not responding to heartbeats. The node is still present in gossip state; Cassandra will attempt to restore communication if heartbeats resume.

LEFT (or "removed") means an operator explicitly decommissioned or removed the node. Cassandra removes it from the ring, redistributes its token ranges, and stops routing to it.

Cassandra never autonomously removes a DOWN node from gossip. This prevents unnecessary data rebalancing during transient failures and avoids the data loss risk of simultaneous range movements.

StorageProxy consults the failure detector via FailureDetector.instance.isAlive(endpoint) before routing requests. A DOWN node is excluded from replica selection for reads; writes are typically directed to hints instead.

IFailureDetector is the interface for FailureDetector, allowing test code to substitute a controlled implementation.

TokenMetadata (org.apache.cassandra.locator) is the in-memory ring view. It maps tokens to endpoints, endpoints to host IDs, and tracks pending operations (bootstrapping nodes, moving nodes). It is the foundation for AbstractReplicationStrategy implementations computing which replicas own a given token.

TokenMetadata is updated as gossip delivers TOKENS and STATUS application state changes.

IEndpointSnitch (org.apache.cassandra.locator) provides rack and datacenter awareness. Implementations translate a node’s IP address into a (datacenter, rack) tuple. NetworkTopologyStrategy uses the snitch to enforce replica placement across failure domains.

When a node joins, leaves, or moves:

In Cassandra 6, Transactional Cluster Metadata (TCM) moves authoritative cluster topology state out of gossip and into a linearizable, log-based metadata service. ClusterMetadata (org.apache.cassandra.tcm) is the single authoritative metadata object for token assignments, keyspace configurations, and node lifecycle state. ClusterMetadataService manages the distributed log and ensures all nodes converge to the same metadata version.

Gossip remains for liveness detection, but schema and topology sources of truth are now ClusterMetadata entries rather than gossip ApplicationState values. See the tcm/ directory in this workspace for TCM-specific documentation drafts.

Streaming is triggered by:

StreamPlan describes the full set of transfers for an operation: which token ranges to send, which tables, and in which direction. StreamResultFuture tracks the overall outcome of all sessions in a plan.

StreamTransferTask handles sending SSTables. There are two transfer modes:

During streaming, SSTables being transferred must remain consistent. LifecycleTransaction ensures SSTables are not compacted away mid-transfer. Streaming failure causes the session to abort and triggers cleanup of partially transferred files. Contributors changing streaming must handle partial transfer and ensure cleanup runs on all failure paths.

For SSTable integrity during streaming, see Compaction and Tombstone Lifecycle.

Repair works by building hierarchical hash trees (Merkle trees) over token ranges. Each leaf in the tree represents a hash of the data in a sub-range. MerkleTrees is constructed per-table per-range during a repair session. Two replicas exchange their trees and identify ranges where the hashes differ.

Full repair hashes the entire dataset for the repaired range. All data, both previously repaired and new, is included in the comparison.

Incremental repair limits comparison to SSTables that have not been marked as repaired. After a successful incremental repair, SSTables are marked as repaired via anti-compaction, and subsequent repairs skip them. This reduces the volume of data compared and transferred during repair.

By default, repair uses range-based Merkle tree comparison, which is efficient for large datasets. Sub-range repair increases tree resolution, allowing finer-grained identification of mismatched ranges (at the cost of more tree construction work).

RepairRunnable is the entry point, triggered by nodetool repair. It constructs a RepairSession per token range, exchanges MerkleTrees with each replica pair, and spawns SyncTask instances for ranges that differ. Sync tasks invoke the streaming layer to transfer the out-of-sync data.

Incremental repair interacts closely with compaction. After repair, AntiCompactionTask splits SSTables into repaired and unrepaired halves. Compaction strategies keep repaired and unrepaired SSTables in separate buckets to avoid mixing them. Mixing repaired and unrepaired SSTables during compaction would invalidate the repaired status and force re-repair of the merged SSTable.

Timeout thresholds are configured in cassandra.yaml and read at runtime via DatabaseDescriptor:

RequestCallback starts a timeout task when a request is dispatched. If the callback has not received enough responses before the timeout fires, a ReadTimeoutException or WriteTimeoutException is returned to the client.

StorageProxy orchestrates retry logic. It uses the failure detector to skip DOWN nodes during replica selection and falls back to hints for writes when replicas are unavailable. There is no automatic client-visible retry at the coordinator level; the exception is returned and the client decides whether to retry.

When a write cannot be delivered to a replica (because it is DOWN), the coordinator stores a hint in the system.hints table. HintedHandoffManager replays hints to the target node when it comes back UP. Hints are best-effort: they are dropped after max_hint_window (default: 3 hours) or when hint storage fills.

Hint delivery is how Cassandra provides eventual consistency without requiring the replica to be present at write time. If a replica is DOWN for longer than max_hint_window, repair is required to restore consistency.