org.apache.hadoop.hbase.replication.regionserver

Class ReplicationSourceManager

java.lang.Object

org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager

All Implemented Interfaces:

ReplicationListener

@InterfaceAudience.Private
public class ReplicationSourceManager
extends Object
implements ReplicationListener

This class is responsible to manage all the replication sources. There are two classes of sources:

• Normal sources are persistent and one per peer cluster
• Old sources are recovered from a failed region server and our only goal is to finish replicating the WAL queue it had

When a region server dies, this class uses a watcher to get notified and it tries to grab a lock in order to transfer all the queues in a local old source.

Synchronization specification:

• No need synchronized on sources. sources is a ConcurrentHashMap and there is a Lock for peer id in PeerProcedureHandlerImpl. So there is no race for peer operations.
• Need synchronized on walsById. There are four methods which modify it, addPeer(String), removePeer(String), cleanOldLogs(NavigableSet, String, boolean, String) and preLogRoll(Path). walsById is a ConcurrentHashMap and there is a Lock for peer id in PeerProcedureHandlerImpl. So there is no race between addPeer(String) and removePeer(String). cleanOldLogs(NavigableSet, String, boolean, String) is called by ReplicationSourceInterface. So no race with addPeer(String). removePeer(String) will terminate the ReplicationSourceInterface firstly, then remove the wals from walsById. So no race with removePeer(String). The only case need synchronized is cleanOldLogs(NavigableSet, String, boolean, String) and preLogRoll(Path).
• No need synchronized on walsByIdRecoveredQueues. There are three methods which modify it, removePeer(String) , cleanOldLogs(NavigableSet, String, boolean, String) and ReplicationSourceManager.NodeFailoverWorker.run(). cleanOldLogs(NavigableSet, String, boolean, String) is called by ReplicationSourceInterface. removePeer(String) will terminate the ReplicationSourceInterface firstly, then remove the wals from walsByIdRecoveredQueues. And ReplicationSourceManager.NodeFailoverWorker.run() will add the wals to walsByIdRecoveredQueues firstly, then start up a ReplicationSourceInterface. So there is no race here. For ReplicationSourceManager.NodeFailoverWorker.run() and removePeer(String), there is already synchronized on oldsources. So no need synchronized on walsByIdRecoveredQueues.
• Need synchronized on latestPaths to avoid the new open source miss new log.
• Need synchronized on oldsources to avoid adding recovered source for the to-be-removed peer.

Constructor Summary

Constructors

Constructor and Description

ReplicationSourceManager(ReplicationQueueStorage queueStorage, ReplicationPeers replicationPeers, ReplicationTracker replicationTracker, org.apache.hadoop.conf.Configuration conf, Server server, org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path logDir, org.apache.hadoop.fs.Path oldLogDir, UUID clusterId, WALFileLengthProvider walFileLengthProvider) Creates a replication manager and sets the watch on all the other registered region servers

Method Summary

Modifier and Type	Method and Description
void	addHFileRefs(TableName tableName, byte[] family, List<Pair<org.apache.hadoop.fs.Path,org.apache.hadoop.fs.Path>> pairs)
void	addPeer(String peerId) 1.
void	cleanUpHFileRefs(String peerId, List<String> files)
org.apache.hadoop.fs.FileSystem	getFs() Get the handle on the local file system
org.apache.hadoop.fs.Path	getLogDir() Get the directory where wals are stored by their RSs
org.apache.hadoop.fs.Path	getOldLogDir() Get the directory where wals are archived
List<ReplicationSourceInterface>	getOldSources() Get a list of all the recovered sources of this rs
ReplicationPeers	getReplicationPeers() Get the ReplicationPeers used by this ReplicationSourceManager
ReplicationSourceInterface	getSource(String peerId) Get the normal source for a given peer
List<ReplicationSourceInterface>	getSources() Get a list of all the normal sources of this rs
String	getStats() Get a string representation of all the sources' metrics
AtomicLong	getTotalBufferUsed()
Map<String,Map<String,NavigableSet<String>>>	getWALs() Get a copy of the wals of the normal sources on this rs
void	join() Terminate the replication on this region server
void	logPositionAndCleanOldLogs(String queueId, boolean queueRecovered, org.apache.hadoop.hbase.replication.regionserver.WALEntryBatch entryBatch) This method will log the current position to storage.
void	postLogRoll(org.apache.hadoop.fs.Path newLog)
void	preLogRoll(org.apache.hadoop.fs.Path newLog)
void	refreshSources(String peerId) Close the previous replication sources of this peer id and open new sources to trigger the new replication state changes or new replication config changes.
void	regionServerRemoved(String regionserver) A region server has been removed from the local cluster
void	removePeer(String peerId) 1.

Methods inherited from class java.lang.Object

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

Constructor Detail

ReplicationSourceManager

public ReplicationSourceManager(ReplicationQueueStorage queueStorage,
                                ReplicationPeers replicationPeers,
                                ReplicationTracker replicationTracker,
                                org.apache.hadoop.conf.Configuration conf,
                                Server server,
                                org.apache.hadoop.fs.FileSystem fs,
                                org.apache.hadoop.fs.Path logDir,
                                org.apache.hadoop.fs.Path oldLogDir,
                                UUID clusterId,
                                WALFileLengthProvider walFileLengthProvider)
                         throws IOException

Creates a replication manager and sets the watch on all the other registered region servers

Parameters:

queueStorage - the interface for manipulating replication queues

replicationPeers -

replicationTracker -

conf - the configuration to use

server - the server for this region server

fs - the file system to use

logDir - the directory that contains all wal directories of live RSs

oldLogDir - the directory where old logs are archived

clusterId -

Throws:

IOException

Method Detail

addPeer

public void addPeer(String peerId)
             throws IOException

1. Add peer to replicationPeers 2. Add the normal source and related replication queue 3. Add HFile Refs

Parameters:

peerId - the id of replication peer

Throws:

IOException

removePeer

public void removePeer(String peerId)

1. Remove peer for replicationPeers 2. Remove all the recovered sources for the specified id and related replication queues 3. Remove the normal source and related replication queue 4. Remove HFile Refs

Parameters:

peerId - the id of the replication peer

refreshSources

public void refreshSources(String peerId)
                    throws IOException

Close the previous replication sources of this peer id and open new sources to trigger the new replication state changes or new replication config changes. Here we don't need to change replication queue storage and only to enqueue all logs to the new replication source

Parameters:

peerId - the id of the replication peer

Throws:

IOException

logPositionAndCleanOldLogs

public void logPositionAndCleanOldLogs(String queueId,
                                       boolean queueRecovered,
                                       org.apache.hadoop.hbase.replication.regionserver.WALEntryBatch entryBatch)

This method will log the current position to storage. And also clean old logs from the replication queue.

Parameters:

queueId - id of the replication queue

queueRecovered - indicates if this queue comes from another region server

entryBatch - the wal entry batch we just shipped

preLogRoll

public void preLogRoll(org.apache.hadoop.fs.Path newLog)
                throws IOException

Throws:

IOException

postLogRoll

public void postLogRoll(org.apache.hadoop.fs.Path newLog)
                 throws IOException

Throws:

IOException

regionServerRemoved

public void regionServerRemoved(String regionserver)

Description copied from interface: ReplicationListener

A region server has been removed from the local cluster

Specified by:

regionServerRemoved in interface ReplicationListener

Parameters:

regionserver - the removed region server

join

public void join()

Terminate the replication on this region server

getWALs

public Map<String,Map<String,NavigableSet<String>>> getWALs()

Get a copy of the wals of the normal sources on this rs

Returns:

a sorted set of wal names

getSources

public List<ReplicationSourceInterface> getSources()

Get a list of all the normal sources of this rs

Returns:

list of all normal sources

getOldSources

public List<ReplicationSourceInterface> getOldSources()

Get a list of all the recovered sources of this rs

Returns:

list of all recovered sources

getSource

public ReplicationSourceInterface getSource(String peerId)

Get the normal source for a given peer

Returns:

the normal source for the give peer if it exists, otherwise null.

getTotalBufferUsed

public AtomicLong getTotalBufferUsed()

getOldLogDir

public org.apache.hadoop.fs.Path getOldLogDir()

Get the directory where wals are archived

Returns:

the directory where wals are archived

getLogDir

public org.apache.hadoop.fs.Path getLogDir()

Get the directory where wals are stored by their RSs

Returns:

the directory where wals are stored by their RSs

getFs

public org.apache.hadoop.fs.FileSystem getFs()

Get the handle on the local file system

Returns:

Handle on the local file system

getReplicationPeers

public ReplicationPeers getReplicationPeers()

Get the ReplicationPeers used by this ReplicationSourceManager

Returns:

the ReplicationPeers used by this ReplicationSourceManager

getStats

public String getStats()

Get a string representation of all the sources' metrics

addHFileRefs

public void addHFileRefs(TableName tableName,
                         byte[] family,
                         List<Pair<org.apache.hadoop.fs.Path,org.apache.hadoop.fs.Path>> pairs)
                  throws IOException

Throws:

IOException

cleanUpHFileRefs

public void cleanUpHFileRefs(String peerId,
                             List<String> files)