Skip to content

Commit

Permalink
HBASE-27516 Document the table based replication queue storage in ref…
Browse files Browse the repository at this point in the history 
… guide
  • Loading branch information
@2005hithlj
2005hithlj committed Apr 26, 2023
1 parent b5535c9 commit 4ad40ab
Showing 1 changed file with 36 additions and 87 deletions.
123 changes: 36 additions & 87 deletions src/main/asciidoc/_chapters/ops_mgt.adoc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2433,26 +2433,22 @@ Replication State Storage::
`ReplicationPeerStorage` and `ReplicationQueueStorage`. The former one is for storing the
replication peer related states, and the latter one is for storing the replication queue related
states.
HBASE-15867 is only half done, as although we have abstract these two interfaces, we still only
have zookeeper based implementations.
And in HBASE-27109, we have implemented the `ReplicationQueueStorage` interface to store the replication queue in the hbase:replication table.

Replication State in ZooKeeper::
By default, the state is contained in the base node _/hbase/replication_.
Usually this nodes contains two child nodes, the `peers` znode is for storing replication peer
state, and the `rs` znodes is for storing replication queue state.
Currently, this nodes contains only one child node, namely `peers` znode, which is used for storing replication peer state.

The `Peers` Znode::
The `peers` znode is stored in _/hbase/replication/peers_ by default.
It consists of a list of all peer replication clusters, along with the status of each of them.
The value of each peer is its cluster key, which is provided in the HBase Shell.
The cluster key contains a list of ZooKeeper nodes in the cluster's quorum, the client port for the ZooKeeper quorum, and the base znode for HBase in HDFS on that cluster.

The `RS` Znode::
The `rs` znode contains a list of WAL logs which need to be replicated.
This list is divided into a set of queues organized by region server and the peer cluster the region server is shipping the logs to.
The rs znode has one child znode for each region server in the cluster.
The child znode name is the region server's hostname, client port, and start code.
This list includes both live and dead region servers.
The `Queue` State::
The `Queue` has been stored in the hbase:replication table, where the row key is <PeerId>-<ServerName>[\t<SourceServerName>], the WAL group will be the qualifier, and the serialized ReplicationGroupOffset will be the value.
The ReplicationGroupOffset includes the wal file of the corresponding queue (<PeerId>-<ServerName>[-<SourceServerName>]) and its offset.
Because we track replication offset per queue instead of per file, we only need to store one replication offset per queue.

Other implementations for `ReplicationPeerStorage`::
Starting from 2.6.0, we introduce a file system based `ReplicationPeerStorage`, which stores
Expand All @@ -2475,14 +2471,14 @@ When nodes are removed from the slave cluster, or if nodes go down or come back

==== Keeping Track of Logs

Each master cluster region server has its own znode in the replication znodes hierarchy.
It contains one znode per peer cluster (if 5 slave clusters, 5 znodes are created), and each of these contain a queue of WALs to process.
Each master cluster region server has its queue state in the hbase:replication table.
It contains one row per peer cluster (if 5 slave clusters, 5 rows are created), and each of these contain a queue of WALs to process.
Each of these queues will track the WALs created by that region server, but they can differ in size.
For example, if one slave cluster becomes unavailable for some time, the WALs should not be deleted, so they need to stay in the queue while the others are processed.
See <<rs.failover.details,rs.failover.details>> for an example.

When a source is instantiated, it contains the current WAL that the region server is writing to.
During log rolling, the new file is added to the queue of each slave cluster's znode just before it is made available.
During log rolling, the new file is added to the queue of each slave cluster's row just before it is made available.
This ensures that all the sources are aware that a new log exists before the region server is able to append edits into it, but this operations is now more expensive.
The queue items are discarded when the replication thread cannot read more entries from a file (because it reached the end of the last block) and there are other files in the queue.
This means that if a source is up to date and replicates from the log that the region server writes to, reading up to the "end" of the current file will not delete the item in the queue.
Expand Down Expand Up @@ -2521,93 +2517,38 @@ NOTE: WALs are saved when replication is enabled or disabled as long as peers ex
[[rs.failover.details]]
==== Region Server Failover

When no region servers are failing, keeping track of the logs in ZooKeeper adds no value.
Unfortunately, region servers do fail, and since ZooKeeper is highly available, it is useful for managing the transfer of the queues in the event of a failure.
When no region servers are failing, keeping track of the logs in hbase:replication table adds no value.
However, in case of region server failure, we will manage the transfer of the queues based on hbase:replication.

Each of the master cluster region servers keeps a watcher on every other region server, in order to be notified when one dies (just as the master does). When a failure happens, they all race to create a znode called `lock` inside the dead region server's znode that contains its queues.
The region server that creates it successfully then transfers all the queues to its own znode, one at a time since ZooKeeper does not support renaming queues.
After queues are all transferred, they are deleted from the old location.
The znodes that were recovered are renamed with the ID of the slave cluster appended with the name of the dead server.
When a region server fails, the HMaster of master cluster will trigger the SCP, and all replication queues on the failed region server will be claimed in the SCP.
The claim queue operation is just to remove the row of a replication queue, and insert a new row, where we change the server name to the region server which claims the queue.

Next, the master cluster region server creates one new source thread per copied queue, and each of the source threads follows the read/filter/ship pattern.
The main difference is that those queues will never receive new data, since they do not belong to their new region server.
When the reader hits the end of the last log, the queue's znode is deleted and the master cluster region server closes that replication source.
When the reader hits the end of the last log, the queue's row is deleted and the master cluster region server closes that replication source.

Given a master cluster with 3 region servers replicating to a single slave with id `2`, the following hierarchy represents what the znodes layout could be at some point in time.
The region servers' znodes all contain a `peers` znode which contains a single queue.
The znode names in the queues represent the actual file names on HDFS in the form `address,port.timestamp`.
Given a master cluster with 3 region servers replicating to a single slave with id `2`, the following info represents what the storage layout of queue in the hbase:replication at some point in time.
Row key is <PeerId>-<ServerName>[\t<SourceServerName>], and value is WAL && Offset.

----
/hbase/replication/rs/
1.1.1.1,60020,123456780/
2/
1.1.1.1,60020.1234 (Contains a position)
1.1.1.1,60020.1265
1.1.1.2,60020,123456790/
2/
1.1.1.2,60020.1214 (Contains a position)
1.1.1.2,60020.1248
1.1.1.2,60020.1312
1.1.1.3,60020, 123456630/
2/
1.1.1.3,60020.1280 (Contains a position)
<PeerId>-<ServerName>[\t<SourceServerName>] WAL && Offset
2-1.1.1.1,60020,123456780 1.1.1.1,60020.1234 (Contains a position)
2-1.1.1.2,60020,123456790 1.1.1.2,60020.1214 (Contains a position)
2-1.1.1.3,60020,123456630 1.1.1.3,60020.1280 (Contains a position)
----

Assume that 1.1.1.2 loses its ZooKeeper session.
The survivors will race to create a lock, and, arbitrarily, 1.1.1.3 wins.
It will then start transferring all the queues to its local peers znode by appending the name of the dead server.
Right before 1.1.1.3 is able to clean up the old znodes, the layout will look like the following:
Assume that 1.1.1.2 failed.
The survivors will claim queue of that, and, arbitrarily, 1.1.1.3 wins.
It will claim all the queue of 1.1.1.2, including removing the row of a replication queue, and inserting a new row(where we change the server name to the region server which claims the queue).
Finally ,the layout will look like the following:

----
/hbase/replication/rs/
1.1.1.1,60020,123456780/
2/
1.1.1.1,60020.1234 (Contains a position)
1.1.1.1,60020.1265
1.1.1.2,60020,123456790/
lock
2/
1.1.1.2,60020.1214 (Contains a position)
1.1.1.2,60020.1248
1.1.1.2,60020.1312
1.1.1.3,60020,123456630/
2/
1.1.1.3,60020.1280 (Contains a position)
2-1.1.1.2,60020,123456790/
1.1.1.2,60020.1214 (Contains a position)
1.1.1.2,60020.1248
1.1.1.2,60020.1312
----

Some time later, but before 1.1.1.3 is able to finish replicating the last WAL from 1.1.1.2, it dies too.
Some new logs were also created in the normal queues.
The last region server will then try to lock 1.1.1.3's znode and will begin transferring all the queues.
The new layout will be:

----
/hbase/replication/rs/
1.1.1.1,60020,123456780/
2/
1.1.1.1,60020.1378 (Contains a position)
2-1.1.1.3,60020,123456630/
1.1.1.3,60020.1325 (Contains a position)
1.1.1.3,60020.1401
2-1.1.1.2,60020,123456790-1.1.1.3,60020,123456630/
1.1.1.2,60020.1312 (Contains a position)
1.1.1.3,60020,123456630/
lock
2/
1.1.1.3,60020.1325 (Contains a position)
1.1.1.3,60020.1401
2-1.1.1.2,60020,123456790/
1.1.1.2,60020.1312 (Contains a position)
<PeerId>-<ServerName>[\t<SourceServerName>] WAL && Offset
2-1.1.1.1,60020,123456780 1.1.1.1,60020.1234 (Contains a position)
2-1.1.1.3,60020,123456630 1.1.1.3,60020.1280 (Contains a position)
2-1.1.1.3,60020,123456630 1.1.1.2,60020,123456790 1.1.1.2,60020.1214 (Contains a position)
----

=== Replication Metrics
Expand Down Expand Up @@ -2694,6 +2635,14 @@ The following metrics are exposed at the global region server level and at the p
| The directory for storing replication peer state, when filesystem replication
peer storage is specified
| peers

| hbase.replication.queue.table.name
| The table for storing replication queue state
| hbase:replication

| hbase.replication.queue.storage.impl
| The replication queue storage implementation
| TableReplicationQueueStorage
|===

=== Monitoring Replication Status
Expand Down

0 comments on commit 4ad40ab

Please sign in to comment.