Package org.apache.hadoop.hbase.replication.regionserver

Class ReplicationSourceManager

java.lang.Object

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

@Private
public class ReplicationSourceManager
extends Object

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 claimQueue(ServerName, String). cleanOldLogs(NavigableSet, String, boolean, String) is called by ReplicationSourceInterface. removePeer(String) will terminate the ReplicationSourceInterface firstly, then remove the wals from walsByIdRecoveredQueues. And claimQueue(ServerName, String) will add the wals to walsByIdRecoveredQueues firstly, then start up a ReplicationSourceInterface. So there is no race here. For claimQueue(ServerName, String) 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	Description
private static interface	ReplicationSourceManager.ReplicationQueueOperation

Field Summary

Fields

Modifier and Type	Field	Description
(package private) AtomicReference<ReplicationSourceInterface>	catalogReplicationSource	A special ReplicationSource for hbase:meta Region Read Replicas.
private final UUID	clusterId
private final org.apache.hadoop.conf.Configuration	conf
private final ThreadPoolExecutor	executor
private final org.apache.hadoop.fs.FileSystem	fs
private final MetricsReplicationGlobalSourceSource	globalMetrics
private final Map<String,org.apache.hadoop.fs.Path>	latestPaths
private static final org.slf4j.Logger	LOG
private final org.apache.hadoop.fs.Path	logDir
private final org.apache.hadoop.fs.Path	oldLogDir
private final List<ReplicationSourceInterface>	oldsources
private final ReplicationQueueStorage	queueStorage	Storage for queues that need persistance; e.g.
private final boolean	replicationForBulkLoadDataEnabled
private final ReplicationPeers	replicationPeers
private final Server	server
private final long	sleepBeforeFailover
private final ConcurrentMap<String,ReplicationSourceInterface>	sources
private final long	totalBufferLimit
private AtomicLong	totalBufferUsed
private final WALFactory	walFactory
private final ConcurrentMap<String,Map<String,NavigableSet<String>>>	walsById
private final ConcurrentMap<String,Map<String,NavigableSet<String>>>	walsByIdRecoveredQueues

Constructor Summary

Constructors

Constructor	Description
ReplicationSourceManager(ReplicationQueueStorage queueStorage, ReplicationPeers replicationPeers, 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, WALFactory walFactory, MetricsReplicationGlobalSourceSource globalMetrics)	Creates a replication manager and sets the watch on all the other registered region servers

Method Summary

Modifier and Type	Method	Description
private void	abortAndThrowIOExceptionWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)
private void	abortWhenFail(ReplicationSourceManager.ReplicationQueueOperation op)
(package private) boolean	acquireBufferQuota(long size)	Add the size to totalBufferUsed and check if it exceeds totalBufferLimit.
(package private) boolean	acquireWALEntryBufferQuota(WALEntryBatch walEntryBatch, WAL.Entry entry)	Acquire the buffer quota for WAL.Entry which is added to WALEntryBatch.
(package private) int	activeFailoverTaskCount()
ReplicationSourceInterface	addCatalogReplicationSource(RegionInfo regionInfo)	Add an hbase:meta Catalog replication source.
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 long	addTotalBufferUsed(long size)
(package private) boolean	checkBufferQuota(String peerId)	Check if totalBufferUsed exceeds totalBufferLimit for peer.
(package private) void	claimQueue(ServerName deadRS, String queue)
(package private) void	cleanOldLogs(String log, boolean inclusive, String queueId, boolean queueRecovered)	Cleans a log file and all older logs from replication queue.
private void	cleanOldLogs(NavigableSet<String> wals, String key, boolean inclusive, String id)
void	cleanUpHFileRefs(String peerId, List<String> files)
private ReplicationSourceInterface	createCatalogReplicationSource(RegionInfo regionInfo)	Create, initialize, and start the Catalog ReplicationSource.
private ReplicationSourceInterface	createSource(String queueId, ReplicationPeer replicationPeer)
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
(package private) MetricsReplicationGlobalSourceSource	getGlobalMetrics()
(package private) Set<org.apache.hadoop.fs.Path>	getLastestPath()
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
(package private) ReplicationQueueStorage	getQueueStorage()
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
long	getTotalBufferLimit()	Returns the maximum size in bytes of edits held in memory which are pending replication across all sources inside this RegionServer.
long	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) void	init()	Adds a normal source per registered peer cluster.
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(ReplicationSourceInterface source, 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.
(package private) void	releaseBufferQuota(long size)	To release the buffer quota which acquired by acquireBufferQuota(long).
(package private) long	releaseWALEntryBatchBufferQuota(WALEntryBatch walEntryBatch)	To release the buffer quota of WALEntryBatch which acquired by acquireWALEntryBufferQuota(org.apache.hadoop.hbase.replication.regionserver.WALEntryBatch, org.apache.hadoop.hbase.wal.WAL.Entry).
void	removeCatalogReplicationSource(RegionInfo regionInfo)	Remove the hbase:meta Catalog replication source.
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)

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

sources

private final ConcurrentMap<String,ReplicationSourceInterface> sources

oldsources

private final List<ReplicationSourceInterface> oldsources

queueStorage

private final ReplicationQueueStorage queueStorage

Storage for queues that need persistance; e.g. Replication state so can be recovered after a crash. queueStorage upkeep is spread about this class and passed to ReplicationSource instances for these to do updates themselves. Not all ReplicationSource instances keep state.

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 Map<String,org.apache.hadoop.fs.Path> latestPaths

logDir

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

oldLogDir

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

walFactory

private final WALFactory walFactory

sleepBeforeFailover

private final long sleepBeforeFailover

executor

private final ThreadPoolExecutor executor

replicationForBulkLoadDataEnabled

private final boolean replicationForBulkLoadDataEnabled

totalBufferUsed

private AtomicLong totalBufferUsed

totalBufferLimit

private final long totalBufferLimit

globalMetrics

private final MetricsReplicationGlobalSourceSource globalMetrics

catalogReplicationSource

AtomicReference<ReplicationSourceInterface> catalogReplicationSource

A special ReplicationSource for hbase:meta Region Read Replicas. Usually this reference remains empty. If an hbase:meta Region is opened on this server, we will create an instance of a hbase:meta CatalogReplicationSource and it will live the life of the Server thereafter; i.e. we will not shut it down even if the hbase:meta moves away from this server (in case it later gets moved back). We synchronize on this instance testing for presence and if absent, while creating so only created and started once.

Constructor Details

ReplicationSourceManager

public ReplicationSourceManager(ReplicationQueueStorage queueStorage,
 ReplicationPeers replicationPeers,
 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,
 WALFactory walFactory,
 MetricsReplicationGlobalSourceSource globalMetrics)
                         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

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

Throws:

IOException

Method Details

init

void init()
   throws IOException

Adds a normal source per registered peer cluster.

Throws:

IOException

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

Parameters:

queueId - the id of the replication queue to associate the ReplicationSource with.

Returns:

a new 'classic' user-space replication source.

Throws:

IOException

See Also:

• for creating a ReplicationSource for meta.

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(ReplicationSourceInterface source,
 WALEntryBatch entryBatch)

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

Parameters:

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

claimQueue

void claimQueue(ServerName deadRS,
 String queue)

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()

getLastestPath

Set<org.apache.hadoop.fs.Path> getLastestPath()

getTotalBufferUsed

public long getTotalBufferUsed()

getTotalBufferLimit

public long getTotalBufferLimit()

Returns the maximum size in bytes of edits held in memory which are pending replication across all sources inside this RegionServer.

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()

getGlobalMetrics

MetricsReplicationGlobalSourceSource getGlobalMetrics()

addCatalogReplicationSource

public ReplicationSourceInterface addCatalogReplicationSource(RegionInfo regionInfo)
                                                       throws IOException

Add an hbase:meta Catalog replication source. Called on open of an hbase:meta Region. Create it once only. If exists already, use the existing one.

Throws:

IOException

See Also:

• removeCatalogReplicationSource(RegionInfo)
• This is specialization on the addSource method.

removeCatalogReplicationSource

public void removeCatalogReplicationSource(RegionInfo regionInfo)

Remove the hbase:meta Catalog replication source. Called when we close hbase:meta.

See Also:

• addCatalogReplicationSource(RegionInfo regionInfo)

createCatalogReplicationSource

private ReplicationSourceInterface createCatalogReplicationSource(RegionInfo regionInfo)
                                                           throws IOException

Create, initialize, and start the Catalog ReplicationSource. Presumes called one-time only (caller must ensure one-time only call). This ReplicationSource is NOT created via ReplicationSourceFactory.

Throws:

IOException

See Also:

• This is a specialization of the addSource call.
• for a note on this ReplicationSource's lifecycle (and more on why the special handling).

getQueueStorage

ReplicationQueueStorage getQueueStorage()

acquireWALEntryBufferQuota

boolean acquireWALEntryBufferQuota(WALEntryBatch walEntryBatch,
 WAL.Entry entry)

Acquire the buffer quota for WAL.Entry which is added to WALEntryBatch.

Parameters:

entry - the wal entry which is added to WALEntryBatch and should acquire buffer quota.

Returns:

true if we should clear buffer and push all

releaseWALEntryBatchBufferQuota

long releaseWALEntryBatchBufferQuota(WALEntryBatch walEntryBatch)

To release the buffer quota of WALEntryBatch which acquired by acquireWALEntryBufferQuota(org.apache.hadoop.hbase.replication.regionserver.WALEntryBatch, org.apache.hadoop.hbase.wal.WAL.Entry).

Returns:

the released buffer quota size.

acquireBufferQuota

boolean acquireBufferQuota(long size)

Add the size to totalBufferUsed and check if it exceeds totalBufferLimit.

Returns:

true if totalBufferUsed exceeds totalBufferLimit,we should stop increase buffer and ship all.

releaseBufferQuota

void releaseBufferQuota(long size)

To release the buffer quota which acquired by acquireBufferQuota(long).

addTotalBufferUsed

private long addTotalBufferUsed(long size)

checkBufferQuota

boolean checkBufferQuota(String peerId)

Check if totalBufferUsed exceeds totalBufferLimit for peer.

Returns:

true if totalBufferUsed not more than totalBufferLimit.