HBase replication architecture — WAL shipping, filters, serial ordering, recovery

What HBase replication actually is

HBase cluster replication is asynchronous log shipping. Every RegionServer already writes each mutation to a write-ahead log for local durability; replication adds a second reader that tails those same WAL files and pushes the edits to remote clusters. Nothing in the client write path waits for the peer: a Put returns once the local WAL and MemStore have it, and the peer sees it milliseconds to minutes later.

That one decision determines everything else. Failover has a non-zero RPO equal to current lag; anything that does not go through the WAL does not replicate by default; and because each RegionServer ships its own log independently, ordering across regions does not exist unless you ask for it.

It is also not a backup: replication copies a bad deleteall to the peer within seconds. Only a point-in-time image protects against operator error - see backup and restore.

The ship path: ReplicationSource to ReplicationEndpoint to sink

Each RegionServer runs a ReplicationSourceManager holding one ReplicationSource per peer. A source owns a queue of WAL file names, an offset into the file it is on, and a reader thread tailing that file - including the live one still being appended to, which is why lag can be sub-second on a healthy link.

The reader pulls WAL.Entry objects - a WALKey plus a WALEdit holding the cells of one row mutation - runs them through a chain of entry filters, and accumulates a batch bounded by entry count and serialized byte size. Shipping is delegated to a ReplicationEndpoint. The default, HBaseInterClusterReplicationEndpoint, resolves the peer's RegionServer list from its ZooKeeper quorum, picks a subset as sinks (a fraction governed by replication.source.ratio), and calls the replicateWALEntry admin RPC on them in parallel. Endpoints are pluggable: a custom class is the supported way to fan WAL edits into Kafka or a search index.

A ReplicationSink converts entries back into ordinary Mutation objects and applies them through the normal write path, so they land in the peer's own WAL and can be forwarded onward in a chained topology. Because every cell keeps its original timestamp, re-applying a batch after a failed RPC converges rather than corrupts: retries are safe.

Peers, cluster keys, and per-peer configuration

A peer is a named destination whose state, cluster key, table scope and bandwidth cap are cluster-wide metadata: every RegionServer picks up changes and spins its sources up or down accordingly.

# peer id, then the target's ZK quorum:port:znode-parent
add_peer '1', CLUSTER_KEY => "zk1,zk2,zk3:2181:/hbase"

# scoped to specific tables / column families, order-preserving
add_peer '2', CLUSTER_KEY => "dr-zk1:2181:/hbase", SERIAL => true,
              TABLE_CFS => { "profiles" => [], "events" => ["cf1"] }

# a column family only replicates if its scope is 1
alter 'profiles', { NAME => 'cf1', REPLICATION_SCOPE => '1' }

set_peer_bandwidth '1', 20971520     # 20 MB/s ceiling for this peer
disable_peer '1'                     # stop shipping, KEEP the queue
status 'replication'                 # per-source lag and queue sizes

disable_peer pauses shipping but retains the queue and the WAL files behind it - the right tool for a planned peer outage. remove_peer deletes the queue permanently: unshipped edits are gone, and re-adding resumes from now, leaving a hole only a snapshot re-seed can fill.

Peer creation and column-family scope are independent switches: a peer with no scoped families ships nothing and reports zero lag while doing so, which on a dashboard looks exactly like a healthy peer.

Scope, filters, and what never leaves the cluster

Selection happens in a chain of WALEntryFilter implementations applied per entry, before batching. ScopeWALEntryFilter drops cells whose column family has REPLICATION_SCOPE = 0, the default, making replication opt-in at column-family rather than table granularity; TableCfWALEntryFilter applies the peer's allow-list; SystemTableWALEntryFilter removes system-table edits, since hbase:meta, ACLs and quota state are cluster-local by construction.

That granularity is the practical lever for data residency: a table with a pii family scoped 0 and a metrics family scoped 1 replicates only the second, from the same rows, with no application change. Delete markers replicate like any other cell - precisely why replication is no protection against accidental deletion - while Increment and Append are logged by their result, not their delta, so shipping them is idempotent.

Bulk loads bypass all of this

Bulk loading writes HFiles straight into the store directory, skipping the WAL, so replication never sees the data. Setting hbase.replication.bulkload.enabled on the source makes the load write a descriptor into the WAL naming the files; the sink then pulls those HFiles from the source cluster's HDFS, which needs network reach, credentials, and a config directory keyed by the source's hbase.replication.cluster.id. The files must still exist when the sink asks, so an archive cleaner TTL shorter than your worst lag produces failures that look like corruption.

Ordering: what parallel replication guarantees, and what serial adds

The default mode preserves order per WAL file, and therefore per RegionServer, but not per region across time. The failure case is specific: a region moves - split, merge, balancer action, RegionServer restart - and its edits now live in two WAL files owned by two servers, each with its own source and shipping speed. The new server can push an edit for that region before the old server has drained edits still queued. Cells carry real timestamps, so the state after both queues drain is correct; it is the intermediate states a peer-side reader sees that go backwards.

Most workloads do not care, because they read the peer only after failover. The ones that do are where a consumer tails the peer and reacts to state transitions: an order that appears SHIPPED before it appears PAID is a real incident even if the row settles a second later.

Serial replication (SERIAL => true) closes the gap with a barrier: when a region opens on a new server HBase records the sequence-id boundary in hbase:meta, and a source holding edits past a barrier will not push them until the recorded push position confirms everything before it has shipped. The cost is real - a region whose predecessor queue is stalled blocks its own edits, so one slow RegionServer holds back regions it used to host. Enable it per peer, only where ordering is a requirement.

Master-master, cyclic topologies, and loop prevention

Nothing stops you pointing two clusters at each other, or building a ring A to B to C to A. What prevents an edit circulating forever is a cluster-id list carried on every WAL key. Each cluster has a UUID; the originator stamps its id on the entry, and a sink applying the mutation attaches the source's id so the peer's own WAL entry inherits the provenance. A source refuses to ship an entry whose id list already contains the destination's cluster id, so an edit stops after one traversal instead of ping-ponging.

Loop prevention is not conflict resolution. The merge rule is last-write-wins on cell timestamp: no vector clocks, no conflict callback. Two clusters writing the same row and qualifier concurrently produce a winner chosen by timestamp, and if the losing write's clock runs a few seconds ahead, the older value wins permanently. Cross-region clock skew is not a theoretical hazard here, it is the failure mode.

Workable master-master designs therefore partition write ownership - by row-key prefix, tenant, or column family - so each cluster is the sole writer for its slice. Non-idempotent operations are the hard stop: Increment and Append run on both sides diverge, because each logs a result computed from a state the other has not seen.

Recovery: what happens when a RegionServer dies

A RegionServer's replication queues - which WAL files remain to ship, and the offset reached in each - are durable metadata, not process state. Historically they live in ZooKeeper under /hbase/replication/rs/<server>/<peerId>/, one znode per WAL file holding the offset; newer versions move that bookkeeping into a system table driven by a Master procedure.

The recovery shape is the same either way. When a server's ephemeral node disappears, surviving RegionServers race for a lock on its queues; the winner adopts them as recovered queues, separate from its own, and drains them from the recorded offsets. That server now ships two streams, which is why one node's death shows up as a lag spike on a different node. If the adopter dies mid-drain, the queue is claimed again.

This is a different mechanism from WAL splitting, which recovers unflushed data for the regions the dead server hosted. Both read the same files, and both must finish before those files can be discarded: ReplicationLogCleaner blocks the archive cleaner from deleting anything a queue still references - the interlock that turns a stalled peer into a disk-space incident rather than data loss.

Lag and the metrics that actually tell you about it

Replication health reduces to a handful of numbers exported per peer per RegionServer. On the source: ageOfLastShippedOp is the age of the most recently shipped edit, the closest thing to an RPO reading; sizeOfLogQueue is the number of WAL files still waiting, which should hover near 1, and any sustained climb means shipping is losing to write throughput; shippedBytes gives the drain rate and failedBatches counts RPCs the peer rejected. On the sink: ageOfLastAppliedOp and appliedOps show the peer absorbing what it accepts.

The trap is that ageOfLastShippedOp measures shipping, not freshness: on a table with no writes it stops moving, so a dead source looks identical to an idle one. Alert on the pair - age above threshold and queue size above 1 - plus a check that the expected sources exist at all. Threshold design is covered in HBase alerting best practices.

For correctness rather than liveness, the VerifyReplication MapReduce job scans both clusters over a time range and reports mismatched, missing and extra rows. Run it over a sampled window rather than the whole table, and note that a TTL or VERSIONS setting differing between clusters produces a permanent, expected mismatch count.

Operational failure modes

oldWALs grows without bound. The most common replication incident by a wide margin. A peer is disabled or hopelessly behind, its queue still references WAL files, the cleaner refuses to delete them, HDFS fills. Not a bug - it is the interlock protecting unshipped edits - and the fix is a decision: repair the peer, or remove it, accept the gap, re-seed from a snapshot.

Schema drift. The peer needs the table with the same name and the same column-family names. Add a family on the source, forget the peer, and every batch containing it fails at the sink, stalling that source completely.

Undersized sink fan-out. A small peer, or a low replication.source.ratio, concentrates inbound traffic on a few RegionServers that become the bottleneck: the peer needs write headroom comparable to the source, not just storage.

Throttles set and forgotten. A bandwidth cap that protects a WAN link in steady state is also why catch-up after an outage takes eleven hours instead of one - treat it as a recovery-time input. For the separate mechanism of RPC and request-level limits, see quota throttling.

Never drilled. Lag graphs at zero prove edits arrive, not that the peer serves traffic.

When to use it, and what to use instead

Cluster replication is the right tool for a continuously warm second cluster with an RPO of seconds, for offloading heavy scans, for migrating with a cutover rather than a freeze, and - via a custom endpoint - as the cheapest live change stream out of HBase.

It is the wrong tool for point-in-time recovery, because it copies mistakes as fast as data: use snapshots or the incremental scheme in backup and restore. It is the wrong tool for read availability inside one data centre, where region replicas serve that need without a second cluster. And it is the wrong tool for synchronous cross-site durability: nothing lets a client wait for the peer, so a zero-RPO requirement calls for a different system. Local durability levels are covered in WAL durability; SKIP_WAL writes are invisible to replication for exactly the reason bulk loads are.