Package org.apache.hadoop.hbase.replication

Class ZKReplicationQueueStorage

java.lang.Object

org.apache.hadoop.hbase.replication.ZKReplicationStorageBase

org.apache.hadoop.hbase.replication.ZKReplicationQueueStorage

All Implemented Interfaces:

ReplicationQueueStorage

@Private
class ZKReplicationQueueStorage
extends ZKReplicationStorageBase
implements ReplicationQueueStorage

ZK based replication queue storage.

The base znode for each regionserver is the regionserver name. For example:

 /hbase/replication/rs/hostname.example.org,6020,1234
 

Within this znode, the region server maintains a set of WAL replication queues. These queues are represented by child znodes named using there give queue id. For example:

 /hbase/replication/rs/hostname.example.org,6020,1234/1
 /hbase/replication/rs/hostname.example.org,6020,1234/2
 

Each queue has one child znode for every WAL that still needs to be replicated. The value of these WAL child znodes is the latest position that has been replicated. This position is updated every time a WAL entry is replicated. For example:

 /hbase/replication/rs/hostname.example.org,6020,1234/1/23522342.23422 [VALUE: 254]
 

Field Summary

Fields

Modifier and Type	Field	Description
private final String	hfileRefsZNode	The name of the znode that contains queues of hfile references to be replicated
private static final org.slf4j.Logger	LOG
private final String	queuesZNode	The name of the znode that contains all replication queues
(package private) final String	regionsZNode
static final String	ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_DEFAULT
static final String	ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_KEY
static final String	ZOOKEEPER_ZNODE_REPLICATION_REGIONS_DEFAULT
static final String	ZOOKEEPER_ZNODE_REPLICATION_REGIONS_KEY

Fields inherited from class org.apache.hadoop.hbase.replication.ZKReplicationStorageBase

conf, REPLICATION_ZNODE, REPLICATION_ZNODE_DEFAULT, replicationZNode, zookeeper

Constructor Summary

Constructors

Constructor	Description
ZKReplicationQueueStorage(ZKWatcher zookeeper, org.apache.hadoop.conf.Configuration conf)

Method Summary

Modifier and Type	Method	Description
void	addHFileRefs(String peerId, List<Pair<org.apache.hadoop.fs.Path,org.apache.hadoop.fs.Path>> pairs)	Add new hfile references to the queue.
private void	addLastSeqIdsToOps(String queueId, Map<String,Long> lastSeqIds, List<ZKUtil.ZKUtilOp> listOfOps)
void	addPeerToHFileRefs(String peerId)	Add a peer to hfile reference queue if peer does not exist.
void	addWAL(ServerName serverName, String queueId, String fileName)	Add a new WAL file to the given queue for a given regionserver.
Pair<String,SortedSet<String>>	claimQueue(ServerName sourceServerName, String queueId, ServerName destServerName)	This implement must update the cversion of root queuesZNode.
Set<String>	getAllHFileRefs()	Load all hfile references in all replication queues.
List<String>	getAllPeersFromHFileRefsQueue()	Get list of all peers from hfile reference queue.
private List<String>	getAllPeersFromHFileRefsQueue0()
List<String>	getAllQueues(ServerName serverName)	Get a list of all queues for the specified region server.
private List<String>	getAllQueues0(ServerName serverName)
Set<String>	getAllWALs()	The optimistic lock of this implement is based on the cversion of root queuesZNode.
private String	getFileNode(String queueNode, String fileName)
private String	getFileNode(ServerName serverName, String queueId, String fileName)
private String	getHFileNode(String peerNode, String fileName)
private String	getHFileRefsPeerNode(String peerId)
protected int	getHFileRefsZNodeCversion()
long	getLastSequenceId(String encodedRegionName, String peerId)	Read the max sequence id of the specific region for a given peer.
protected Pair<Long,Integer>	getLastSequenceIdWithVersion(String encodedRegionName, String peerId)	Return the {lastPushedSequenceId, ZNodeDataVersion} pair.
List<ServerName>	getListOfReplicators()	Get a list of all region servers that have outstanding replication queues.
private List<ServerName>	getListOfReplicators0()
private String	getQueueNode(ServerName serverName, String queueId)
protected int	getQueuesZNodeCversion()
List<String>	getReplicableHFiles(String peerId)	Get a list of all hfile references in the given peer.
private List<String>	getReplicableHFiles0(String peerId)
String	getRsNode(ServerName serverName)	Get full znode name for given region server
(package private) String	getSerialReplicationRegionPeerNode(String encodedRegionName, String peerId)	Put all regions under /hbase/replication/regions znode will lead to too many children because of the huge number of regions in real production environment.
long	getWALPosition(ServerName serverName, String queueId, String fileName)	Get the current position for a specific WAL in a given queue for a given regionserver.
List<String>	getWALsInQueue(ServerName serverName, String queueId)	Get a list of all WALs in the given queue on the given region server.
private List<String>	getWALsInQueue0(ServerName serverName, String queueId)
void	removeHFileRefs(String peerId, List<String> files)	Remove hfile references from the queue.
void	removeLastSequenceIds(String peerId)	Remove all the max sequence id record for the given peer.
void	removeLastSequenceIds(String peerId, List<String> encodedRegionNames)	Remove the max sequence id record for the given peer and regions.
void	removePeerFromHFileRefs(String peerId)	Remove a peer from hfile reference queue.
void	removeQueue(ServerName serverName, String queueId)	Remove a replication queue for a given regionserver.
void	removeReplicatorIfQueueIsEmpty(ServerName serverName)	Remove the record of region server if the queue is empty.
void	removeWAL(ServerName serverName, String queueId, String fileName)	Remove an WAL file from the given queue for a given regionserver.
void	setLastSequenceIds(String peerId, Map<String,Long> lastSeqIds)	Set the max sequence id of a bunch of regions for a given peer.
void	setWALPosition(ServerName serverName, String queueId, String fileName, long position, Map<String,Long> lastSeqIds)	Set the current position for a specific WAL in a given queue for a given regionserver.

Methods inherited from class org.apache.hadoop.hbase.replication.ZKReplicationStorageBase

toByteArray

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

LOG

private static final org.slf4j.Logger LOG

ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_KEY

public static final String ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_KEY

See Also:

• Constant Field Values

ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_DEFAULT

public static final String ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_DEFAULT

See Also:

• Constant Field Values

ZOOKEEPER_ZNODE_REPLICATION_REGIONS_KEY

public static final String ZOOKEEPER_ZNODE_REPLICATION_REGIONS_KEY

See Also:

• Constant Field Values

ZOOKEEPER_ZNODE_REPLICATION_REGIONS_DEFAULT

public static final String ZOOKEEPER_ZNODE_REPLICATION_REGIONS_DEFAULT

See Also:

• Constant Field Values

queuesZNode

private final String queuesZNode

The name of the znode that contains all replication queues

hfileRefsZNode

private final String hfileRefsZNode

The name of the znode that contains queues of hfile references to be replicated

regionsZNode

final String regionsZNode

Constructor Details

ZKReplicationQueueStorage

public ZKReplicationQueueStorage(ZKWatcher zookeeper,
 org.apache.hadoop.conf.Configuration conf)

Method Details

getRsNode

public String getRsNode(ServerName serverName)

Description copied from interface: ReplicationQueueStorage

Get full znode name for given region server

Specified by:

getRsNode in interface ReplicationQueueStorage

Parameters:

serverName - the name of the region server

Returns:

full znode name

getQueueNode

private String getQueueNode(ServerName serverName,
 String queueId)

getFileNode

private String getFileNode(String queueNode,
 String fileName)

getFileNode

private String getFileNode(ServerName serverName,
 String queueId,
 String fileName)

getSerialReplicationRegionPeerNode

String getSerialReplicationRegionPeerNode(String encodedRegionName,
 String peerId)

Put all regions under /hbase/replication/regions znode will lead to too many children because of the huge number of regions in real production environment. So here we will distribute the znodes to multiple directories.

So the final znode path will be format like this:

 /hbase/replication/regions/dd/04/e76a6966d4ffa908ed0586764767-100
 

Here the full encoded region name is dd04e76a6966d4ffa908ed0586764767, and we use the first two characters 'dd' as the first level directory name, and use the next two characters '04' as the second level directory name, and the rest part as the prefix of the znode, and the suffix '100' is the peer id.

Parameters:

encodedRegionName - the encoded region name.

peerId - peer id for replication.

Returns:

ZNode path to persist the max sequence id that we've pushed for the given region and peer.

removeQueue

public void removeQueue(ServerName serverName,
 String queueId)
                 throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Remove a replication queue for a given regionserver.

Specified by:

removeQueue in interface ReplicationQueueStorage

Parameters:

serverName - the name of the regionserver

queueId - a String that identifies the queue.

Throws:

ReplicationException

addWAL

public void addWAL(ServerName serverName,
 String queueId,
 String fileName)
            throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Add a new WAL file to the given queue for a given regionserver. If the queue does not exist it is created.

Specified by:

addWAL in interface ReplicationQueueStorage

Parameters:

serverName - the name of the regionserver

queueId - a String that identifies the queue.

fileName - name of the WAL

Throws:

ReplicationException

removeWAL

public void removeWAL(ServerName serverName,
 String queueId,
 String fileName)
               throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Remove an WAL file from the given queue for a given regionserver.

Specified by:

removeWAL in interface ReplicationQueueStorage

Parameters:

serverName - the name of the regionserver

queueId - a String that identifies the queue.

fileName - name of the WAL

Throws:

ReplicationException

addLastSeqIdsToOps

private void addLastSeqIdsToOps(String queueId,
 Map<String,Long> lastSeqIds,
 List<ZKUtil.ZKUtilOp> listOfOps)
                         throws org.apache.zookeeper.KeeperException,
ReplicationException

Throws:

org.apache.zookeeper.KeeperException

ReplicationException

setWALPosition

public void setWALPosition(ServerName serverName,
 String queueId,
 String fileName,
 long position,
 Map<String,Long> lastSeqIds)
                    throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Set the current position for a specific WAL in a given queue for a given regionserver.

Specified by:

setWALPosition in interface ReplicationQueueStorage

Parameters:

serverName - the name of the regionserver

queueId - a String that identifies the queue

fileName - name of the WAL

position - the current position in the file. Will ignore if less than or equal to 0.

lastSeqIds - map with {encodedRegionName, sequenceId} pairs for serial replication.

Throws:

ReplicationException

getLastSequenceIdWithVersion

protected Pair<Long,Integer> getLastSequenceIdWithVersion(String encodedRegionName,
 String peerId)
                                                   throws org.apache.zookeeper.KeeperException

Return the {lastPushedSequenceId, ZNodeDataVersion} pair. if ZNodeDataVersion is -1, it means that the ZNode does not exist.

Throws:

org.apache.zookeeper.KeeperException

getLastSequenceId

public long getLastSequenceId(String encodedRegionName,
 String peerId)
                       throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Read the max sequence id of the specific region for a given peer. For serial replication, we need the max sequenced id to decide whether we can push the next entries.

Specified by:

getLastSequenceId in interface ReplicationQueueStorage

Parameters:

encodedRegionName - the encoded region name

peerId - peer id

Returns:

the max sequence id of the specific region for a given peer.

Throws:

ReplicationException

setLastSequenceIds

public void setLastSequenceIds(String peerId,
 Map<String,Long> lastSeqIds)
                        throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Set the max sequence id of a bunch of regions for a given peer. Will be called when setting up a serial replication peer.

Specified by:

setLastSequenceIds in interface ReplicationQueueStorage

Parameters:

peerId - peer id

lastSeqIds - map with {encodedRegionName, sequenceId} pairs for serial replication.

Throws:

ReplicationException

removeLastSequenceIds

public void removeLastSequenceIds(String peerId)
                           throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Remove all the max sequence id record for the given peer.

Specified by:

removeLastSequenceIds in interface ReplicationQueueStorage

Parameters:

peerId - peer id

Throws:

ReplicationException

removeLastSequenceIds

public void removeLastSequenceIds(String peerId,
 List<String> encodedRegionNames)
                           throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Remove the max sequence id record for the given peer and regions.

Specified by:

removeLastSequenceIds in interface ReplicationQueueStorage

Parameters:

peerId - peer id

encodedRegionNames - the encoded region names

Throws:

ReplicationException

getWALPosition

public long getWALPosition(ServerName serverName,
 String queueId,
 String fileName)
                    throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Get the current position for a specific WAL in a given queue for a given regionserver.

Specified by:

getWALPosition in interface ReplicationQueueStorage

Parameters:

serverName - the name of the regionserver

queueId - a String that identifies the queue

fileName - name of the WAL

Returns:

the current position in the file

Throws:

ReplicationException

claimQueue

public Pair<String,SortedSet<String>> claimQueue(ServerName sourceServerName,
 String queueId,
 ServerName destServerName)
                                          throws ReplicationException

This implement must update the cversion of root queuesZNode. The optimistic lock of the getAllWALs() method is based on the cversion of root queuesZNode.

Specified by:

claimQueue in interface ReplicationQueueStorage

Parameters:

sourceServerName - the name of the dead region server

queueId - the id of the queue

destServerName - the name of the target region server

Returns:

the new PeerId and A SortedSet of WALs in its queue

Throws:

ReplicationException

See Also:

• to show the usage of the cversion of root #queuesZNode .

removeReplicatorIfQueueIsEmpty

public void removeReplicatorIfQueueIsEmpty(ServerName serverName)
                                    throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Remove the record of region server if the queue is empty.

Specified by:

removeReplicatorIfQueueIsEmpty in interface ReplicationQueueStorage

Throws:

ReplicationException

getListOfReplicators0

private List<ServerName> getListOfReplicators0()
                                        throws org.apache.zookeeper.KeeperException

Throws:

org.apache.zookeeper.KeeperException

getListOfReplicators

public List<ServerName> getListOfReplicators()
                                      throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Get a list of all region servers that have outstanding replication queues. These servers could be alive, dead or from a previous run of the cluster.

Specified by:

getListOfReplicators in interface ReplicationQueueStorage

Returns:

a list of server names

Throws:

ReplicationException

getWALsInQueue0

private List<String> getWALsInQueue0(ServerName serverName,
 String queueId)
                              throws org.apache.zookeeper.KeeperException

Throws:

org.apache.zookeeper.KeeperException

getWALsInQueue

public List<String> getWALsInQueue(ServerName serverName,
 String queueId)
                            throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Get a list of all WALs in the given queue on the given region server.

Specified by:

getWALsInQueue in interface ReplicationQueueStorage

Parameters:

serverName - the server name of the region server that owns the queue

queueId - a String that identifies the queue

Returns:

a list of WALs

Throws:

ReplicationException

getAllQueues0

private List<String> getAllQueues0(ServerName serverName)
                            throws org.apache.zookeeper.KeeperException

Throws:

org.apache.zookeeper.KeeperException

getAllQueues

public List<String> getAllQueues(ServerName serverName)
                          throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Get a list of all queues for the specified region server.

Specified by:

getAllQueues in interface ReplicationQueueStorage

Parameters:

serverName - the server name of the region server that owns the set of queues

Returns:

a list of queueIds

Throws:

ReplicationException

getQueuesZNodeCversion

protected int getQueuesZNodeCversion()
                              throws org.apache.zookeeper.KeeperException

Throws:

org.apache.zookeeper.KeeperException

getAllWALs

public Set<String> getAllWALs()
                       throws ReplicationException

The optimistic lock of this implement is based on the cversion of root queuesZNode. Therefore, we must update the cversion of root queuesZNode when migrate wal nodes to other queues.

Specified by:

getAllWALs in interface ReplicationQueueStorage

Throws:

ReplicationException

See Also:

• as an example of updating root #queuesZNode cversion.

getHFileRefsPeerNode

private String getHFileRefsPeerNode(String peerId)

getHFileNode

private String getHFileNode(String peerNode,
 String fileName)

addPeerToHFileRefs

public void addPeerToHFileRefs(String peerId)
                        throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Add a peer to hfile reference queue if peer does not exist.

Specified by:

addPeerToHFileRefs in interface ReplicationQueueStorage

Parameters:

peerId - peer cluster id to be added

Throws:

ReplicationException - if fails to add a peer id to hfile reference queue

removePeerFromHFileRefs

public void removePeerFromHFileRefs(String peerId)
                             throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Remove a peer from hfile reference queue.

Specified by:

removePeerFromHFileRefs in interface ReplicationQueueStorage

Parameters:

peerId - peer cluster id to be removed

Throws:

ReplicationException

addHFileRefs

public void addHFileRefs(String peerId,
 List<Pair<org.apache.hadoop.fs.Path,org.apache.hadoop.fs.Path>> pairs)
                  throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Add new hfile references to the queue.

Specified by:

addHFileRefs in interface ReplicationQueueStorage

Parameters:

peerId - peer cluster id to which the hfiles need to be replicated

pairs - list of pairs of { HFile location in staging dir, HFile path in region dir which will be added in the queue }

Throws:

ReplicationException - if fails to add a hfile reference

removeHFileRefs

public void removeHFileRefs(String peerId,
 List<String> files)
                     throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Remove hfile references from the queue.

Specified by:

removeHFileRefs in interface ReplicationQueueStorage

Parameters:

peerId - peer cluster id from which this hfile references needs to be removed

files - list of hfile references to be removed

Throws:

ReplicationException

getAllPeersFromHFileRefsQueue0

private List<String> getAllPeersFromHFileRefsQueue0()
                                             throws org.apache.zookeeper.KeeperException

Throws:

org.apache.zookeeper.KeeperException

getAllPeersFromHFileRefsQueue

public List<String> getAllPeersFromHFileRefsQueue()
                                           throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Get list of all peers from hfile reference queue.

Specified by:

getAllPeersFromHFileRefsQueue in interface ReplicationQueueStorage

Returns:

a list of peer ids

Throws:

ReplicationException

getReplicableHFiles0

private List<String> getReplicableHFiles0(String peerId)
                                   throws org.apache.zookeeper.KeeperException

Throws:

org.apache.zookeeper.KeeperException

getReplicableHFiles

public List<String> getReplicableHFiles(String peerId)
                                 throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Get a list of all hfile references in the given peer.

Specified by:

getReplicableHFiles in interface ReplicationQueueStorage

Parameters:

peerId - a String that identifies the peer

Returns:

a list of hfile references

Throws:

ReplicationException

getHFileRefsZNodeCversion

protected int getHFileRefsZNodeCversion()
                                 throws ReplicationException

Throws:

ReplicationException

getAllHFileRefs

public Set<String> getAllHFileRefs()
                            throws ReplicationException

Description copied from interface: ReplicationQueueStorage

Load all hfile references in all replication queues. This method guarantees to return a snapshot which contains all hfile references at the start of this call. However, some newly created hfile references during the call may not be included.

Specified by:

getAllHFileRefs in interface ReplicationQueueStorage

Throws:

ReplicationException