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.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
(package private) class	ReplicationSourceManager.NodeFailoverWorker Class responsible to setup new ReplicationSources to take care of the queues from dead region servers.
private static interface	ReplicationSourceManager.ReplicationQueueOperation

Field Summary

Fields

Modifier and Type	Field and Description
private UUID	clusterId
private org.apache.hadoop.conf.Configuration	conf
private ThreadPoolExecutor	executor
private org.apache.hadoop.fs.FileSystem	fs
private Set<org.apache.hadoop.fs.Path>	latestPaths
private static org.slf4j.Logger	LOG
private org.apache.hadoop.fs.Path	logDir
private org.apache.hadoop.fs.Path	oldLogDir
private List<ReplicationSourceInterface>	oldsources
private ReplicationQueueStorage	queueStorage
private boolean	replicationForBulkLoadDataEnabled
private ReplicationPeers	replicationPeers
private ReplicationTracker	replicationTracker
private Server	server
private long	sleepBeforeFailover
private ConcurrentMap<String,ReplicationSourceInterface>	sources
private AtomicLong	totalBufferUsed
private WALFileLengthProvider	walFileLengthProvider
private ConcurrentMap<String,Map<String,NavigableSet<String>>>	walsById
private ConcurrentMap<String,Map<String,NavigableSet<String>>>	walsByIdRecoveredQueues

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
private void	abortAndThrowIOExceptionWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)
private void	abortWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)
(package private) int	activeFailoverTaskCount()
void	addHFileRefs(TableName tableName, byte[] family, List<Pair<org.apache.hadoop.fs.Path,org.apache.hadoop.fs.Path>> pairs)
void	addPeer(String peerId) 1.
(package private) ReplicationSourceInterface	addSource(String peerId) Add a normal source for the given peer on this region server.
private void	adoptAbandonedQueues()
private void	cleanOldLogs(NavigableSet<String> wals, String key, boolean inclusive, String id)
(package private) void	cleanOldLogs(String log, boolean inclusive, String queueId, boolean queueRecovered) Cleans a log file and all older logs from replication queue.
void	cleanUpHFileRefs(String peerId, List<String> files)
private ReplicationSourceInterface	createSource(String queueId, ReplicationPeer replicationPeer) Factory method to create a replication source
private void	deleteQueue(String queueId) Delete a complete queue of wals associated with a replication source
(package private) List<String>	getAllQueues()
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
(package private) int	getSizeOfLatestPath()
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
(package private) Map<String,Map<String,NavigableSet<String>>>	getWalsByIdRecoveredQueues() Get a copy of the wals of the recovered sources on this rs
(package private) Future<?>	init() Adds a normal source per registered peer cluster and tries to process all old region server wal queues
private void	interruptOrAbortWhenFail(ReplicationSourceManager.ReplicationQueueOperation op) Refresh replication source will terminate the old source first, then the source thread will be interrupted.
void	join() Terminate the replication on this region server
void	logPositionAndCleanOldLogs(String queueId, boolean queueRecovered, 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.
(package private) void	removeRecoveredSource(ReplicationSourceInterface src) Clear the metrics and related replication queue of the specified old source
(package private) void	removeSource(ReplicationSourceInterface src) Clear the metrics and related replication queue of the specified old source
private void	throwIOExceptionWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)
private void	transferQueues(ServerName deadRS) Transfer all the queues of the specified to this region server.

Methods inherited from class java.lang.Object

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

Field Detail

LOG

private static final org.slf4j.Logger LOG

sources

private final ConcurrentMap<String,ReplicationSourceInterface> sources

oldsources

private final List<ReplicationSourceInterface> oldsources

queueStorage

private final ReplicationQueueStorage queueStorage

replicationTracker

private final ReplicationTracker replicationTracker

replicationPeers

private final ReplicationPeers replicationPeers

clusterId

private final UUID clusterId

server

private final Server server

walsById

private final ConcurrentMap<String,Map<String,NavigableSet<String>>> walsById

walsByIdRecoveredQueues

private final ConcurrentMap<String,Map<String,NavigableSet<String>>> walsByIdRecoveredQueues

conf

private final org.apache.hadoop.conf.Configuration conf

fs

private final org.apache.hadoop.fs.FileSystem fs

latestPaths

private final Set<org.apache.hadoop.fs.Path> latestPaths

logDir

private final org.apache.hadoop.fs.Path logDir

oldLogDir

private final org.apache.hadoop.fs.Path oldLogDir

walFileLengthProvider

private final WALFileLengthProvider walFileLengthProvider

sleepBeforeFailover

private final long sleepBeforeFailover

executor

private final ThreadPoolExecutor executor

replicationForBulkLoadDataEnabled

private final boolean replicationForBulkLoadDataEnabled

totalBufferUsed

private AtomicLong totalBufferUsed

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

init

Future<?> init()
        throws IOException

Adds a normal source per registered peer cluster and tries to process all old region server wal queues

The returned future is for adoptAbandonedQueues task.

Throws:

IOException

adoptAbandonedQueues

private void adoptAbandonedQueues()

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

createSource

private ReplicationSourceInterface createSource(String queueId,
                                                ReplicationPeer replicationPeer)
                                         throws IOException

Factory method to create a replication source

Parameters:

queueId - the id of the replication queue

Returns:

the created source

Throws:

IOException

addSource

ReplicationSourceInterface addSource(String peerId)
                              throws IOException

Add a normal source for the given peer on this region server. Meanwhile, add new replication queue to storage. For the newly added peer, we only need to enqueue the latest log of each wal group and do replication

Parameters:

peerId - the id of the replication peer

Returns:

the source that was created

Throws:

IOException

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

removeRecoveredSource

void removeRecoveredSource(ReplicationSourceInterface src)

Clear the metrics and related replication queue of the specified old source

Parameters:

src - source to clear

removeSource

void removeSource(ReplicationSourceInterface src)

Clear the metrics and related replication queue of the specified old source

Parameters:

src - source to clear

deleteQueue

private void deleteQueue(String queueId)

Delete a complete queue of wals associated with a replication source

Parameters:

queueId - the id of replication queue to delete

interruptOrAbortWhenFail

private void interruptOrAbortWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)

Refresh replication source will terminate the old source first, then the source thread will be interrupted. Need to handle it instead of abort the region server.

abortWhenFail

private void abortWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)

throwIOExceptionWhenFail

private void throwIOExceptionWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)
                               throws IOException

Throws:

IOException

abortAndThrowIOExceptionWhenFail

private void abortAndThrowIOExceptionWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)
                                       throws IOException

Throws:

IOException

logPositionAndCleanOldLogs

public void logPositionAndCleanOldLogs(String queueId,
                                       boolean queueRecovered,
                                       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

cleanOldLogs

void cleanOldLogs(String log,
                  boolean inclusive,
                  String queueId,
                  boolean queueRecovered)

Cleans a log file and all older logs from replication queue. Called when we are sure that a log file is closed and has no more entries.

Parameters:

log - Path to the log

inclusive - whether we should also remove the given log file

queueId - id of the replication queue

queueRecovered - Whether this is a recovered queue

cleanOldLogs

private void cleanOldLogs(NavigableSet<String> wals,
                          String key,
                          boolean inclusive,
                          String id)

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

transferQueues

private void transferQueues(ServerName deadRS)

Transfer all the queues of the specified to this region server. First it tries to grab a lock and if it works it will move the old queues and finally will delete the old queues.

It creates one old source for any type of source of the old rs.

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

getWalsByIdRecoveredQueues

Map<String,Map<String,NavigableSet<String>>> getWalsByIdRecoveredQueues()

Get a copy of the wals of the recovered 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.

getAllQueues

List<String> getAllQueues()
                   throws IOException

Throws:

IOException

getSizeOfLatestPath

int getSizeOfLatestPath()

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)

activeFailoverTaskCount

int activeFailoverTaskCount()