HBASE-27516 Document the table based replication queue storage in ref…

… guide

src/main/asciidoc/_chapters/ops_mgt.adoc

@@ -2429,17 +2429,13 @@ This option was introduced in link:https://issues.apache.org/jira/browse/HBASE-1
 ==== Replication Internals
 
 Replication State Storage::
-  In HBASE-15867, we abstract two interfaces for storing replication state,
-`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-27110, we have implemented a file system based replication peer storage, to store replication peer state on file system. Of course you can still use the zookeeper based replication peer storage.
+And in HBASE-27109, we have changed the replication queue storage from zookeeper based to hbase table based. See the below `Replication Queue State` in hbase:replication table section for more details.
 
 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.
+  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. And if you choose the file system based replication peer storage, you will not see the peers znode.
+  And starting from 3.0.0, we have moved the replication queue state to <<hbase:replication,hbase:replication>> table, so you will not see the rs znode.
 
 The `Peers` Znode::
   The `peers` znode is stored in _/hbase/replication/peers_ by default.
@@ -2454,6 +2450,12 @@ The `RS` Znode::
   The child znode name is the region server's hostname, client port, and start code.
   This list includes both live and dead region servers.
 
+[[hbase:replication]]
+hbase:replication::
+  After 3.0.0, the `Queue` has been stored in the hbase:replication table, where the row key is <PeerId>-<ServerName>[/<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
 the replication peer state with files on HFile file system, instead of znodes on ZooKeeper.
@@ -2473,7 +2475,7 @@ A ZooKeeper watcher is placed on the _${zookeeper.znode.parent}/rs_ node of the
 This watch is used to monitor changes in the composition of the slave cluster.
 When nodes are removed from the slave cluster, or if nodes go down or come back up, the master cluster's region servers will respond by selecting a new pool of slave region servers to replicate to.
 
-==== Keeping Track of Logs
+==== Keeping Track of Logs(based on ZooKeeper)
 
 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.
@@ -2494,6 +2496,18 @@ If the log is in the queue, the path will be updated in memory.
 If the log is currently being replicated, the change will be done atomically so that the reader doesn't attempt to open the file when has already been moved.
 Because moving a file is a NameNode operation , if the reader is currently reading the log, it won't generate any exception.
 
+==== Keeping Track of Logs(based on hbase table)
+
+After 3.0.0, for table based implementation, we have server name in row key, which means we will have lots of rows for a given peer.
+
+For a normal replication queue, where the WAL files belong to it is still alive, all the WAL files are kept in memory, so we do not need to get the WAL files from replication queue storage.
+And for a recovered replication queue, we could get the WAL files of the dead region server by listing the old WAL directory on HDFS. So theoretically, we do not need to store every WAL file in replication queue storage.
+And what’s more, we store the created time(usually) in the WAL file name, so for all the WAL files in a WAL group, we can sort them(actually we will sort them in the current replication framework), which means we only need to store one replication offset per queue.
+When starting a recovered replication queue, we will skip all the files before this offset, and start replicating from this offset.
+
+For ReplicationLogCleaner, all the files before this offset can be deleted, otherwise not.
+
+
 ==== Reading, Filtering and Sending Edits
 
 By default, a source attempts to read from a WAL and ship log entries to a sink as quickly as possible.
@@ -2519,12 +2533,12 @@ The next time the cleaning process needs to look for a log, it starts by using i
 NOTE: WALs are saved when replication is enabled or disabled as long as peers exist.
 
 [[rs.failover.details]]
-==== Region Server Failover
+==== Region Server Failover(based on ZooKeeper)
 
 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.
-
-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.
+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.
@@ -2587,27 +2601,38 @@ 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:
 
+==== Region Server Failover(based on hbase table)
+
+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.
+Here, we use multi row mutate endpoint atomically, so the data for a single peer must be in the same region.
+
+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 record 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 info represents what the storage layout of queue in the hbase:replication at some point in time.
+Row key is <PeerId>-<ServerName>[/<SourceServerName>], and value is WAL && Offset.
+
 ----
 
-/hbase/replication/rs/
-  1.1.1.1,60020,123456780/
-    2/
-      1.1.1.1,60020.1378  (Contains a position)
+  <PeerId>-<ServerName>[/<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)
+----
 
-    2-1.1.1.3,60020,123456630/
-      1.1.1.3,60020.1325  (Contains a position)
-      1.1.1.3,60020.1401
+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:
 
-    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>[/<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
@@ -2694,6 +2719,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