org.apache.hadoop.hbase.replication.regionserver

Class ReplicationSourceManager

java.lang.Object

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

@InterfaceAudience.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(String, boolean, ReplicationSourceInterface) and postLogRoll(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(String, boolean, ReplicationSourceInterface) 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(String, boolean, ReplicationSourceInterface) and postLogRoll(Path).
• No need synchronized on walsByIdRecoveredQueues. There are three methods which modify it, removePeer(String) , cleanOldLogs(String, boolean, ReplicationSourceInterface) and claimQueue(ReplicationQueueId). cleanOldLogs(String, boolean, ReplicationSourceInterface) is called by ReplicationSourceInterface. removePeer(String) will terminate the ReplicationSourceInterface firstly, then remove the wals from walsByIdRecoveredQueues. And claimQueue(ReplicationQueueId) will add the wals to walsByIdRecoveredQueues firstly, then start up a ReplicationSourceInterface. So there is no race here. For claimQueue(ReplicationQueueId) 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
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 MetricsReplicationGlobalSourceSource	globalMetrics
private Map<String,org.apache.hadoop.fs.Path>	latestPaths
private static org.slf4j.Logger	LOG
private org.apache.hadoop.fs.Path	logDir
private int	maxRetriesMultiplier
private org.apache.hadoop.fs.Path	oldLogDir
private List<ReplicationSourceInterface>	oldsources
private ReplicationQueueStorage	queueStorage Storage for queues that need persistance; e.g.
private ReplicationPeers	replicationPeers
private Server	server
private long	sleepBeforeFailover
private long	sleepForRetries
private ConcurrentMap<String,ReplicationSourceInterface>	sources
private SyncReplicationPeerMappingManager	syncReplicationPeerMappingManager
private long	totalBufferLimit
private AtomicLong	totalBufferUsed
private WALFactory	walFactory
private ConcurrentMap<ReplicationQueueId,Map<String,NavigableSet<String>>>	walsById
private ConcurrentMap<ReplicationQueueId,Map<String,NavigableSet<org.apache.hadoop.fs.Path>>>	walsByIdRecoveredQueues

Constructor Summary

Constructors

Constructor and 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, SyncReplicationPeerMappingManager syncReplicationPeerMappingManager, MetricsReplicationGlobalSourceSource globalMetrics) 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) 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()
void	addHFileRefs(TableName tableName, byte[] family, List<Pair<org.apache.hadoop.fs.Path,org.apache.hadoop.fs.Path>> pairs)
void	addPeer(String peerId) Add peer to replicationPeers Add the normal source and related replication queue Add HFile Refs
private void	addRecoveredSource(ReplicationSourceInterface src, ReplicationPeerImpl oldPeer, ReplicationQueueId claimedQueueId, PriorityQueue<org.apache.hadoop.fs.Path> walFiles)
(package private) void	addSource(String peerId, boolean init) 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(ReplicationQueueId queueId)
(package private) void	claimQueue(ReplicationQueueId queueId, boolean syncUp) Claim a replication queue.
private void	cleanOldLogs(NavigableSet<String> wals, ReplicationSourceInterface source)
(package private) void	cleanOldLogs(String log, boolean inclusive, ReplicationSourceInterface source) Cleans a log file and all older logs from replication queue.
void	cleanUpHFileRefs(String peerId, List<String> files)
private ReplicationSourceInterface	createRefreshedSource(ReplicationQueueId queueId, ReplicationPeer peer)
private ReplicationSourceInterface	createSource(ReplicationQueueData queueData, ReplicationPeer replicationPeer)
private void	deleteQueue(ReplicationQueueId queueId) Delete a complete queue of wals associated with a replication source
void	drainSources(String peerId) This is used when we transit a sync replication peer to SyncReplicationState.STANDBY.
(package private) void	finishRecoveredSource(ReplicationSourceInterface src)
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()
private PriorityQueue<org.apache.hadoop.fs.Path>	getWALFilesToReplicate(ServerName sourceRS, boolean syncUp, Map<String,ReplicationGroupOffset> offsets)
Map<ReplicationQueueId,Map<String,NavigableSet<String>>>	getWALs() Get a copy of the wals of the normal 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	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	removePeer(String peerId) Remove peer for replicationPeers Remove all the recovered sources for the specified id and related replication queues Remove the normal source and related replication queue Remove HFile Refs
private boolean	removeRecoveredSource(ReplicationSourceInterface src) Clear the metrics and related replication queue of the specified old source
private void	removeRemoteWALs(String peerId, String remoteWALDir, Collection<String> wals)
(package private) void	removeSource(ReplicationSourceInterface src) Clear the metrics and related replication queue of the specified old source
private boolean	shouldReplicate(ReplicationGroupOffset offset, String wal) Check whether we should replicate the given wal.
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 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

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<ReplicationQueueId,Map<String,NavigableSet<String>>> walsById

walsByIdRecoveredQueues

private final ConcurrentMap<ReplicationQueueId,Map<String,NavigableSet<org.apache.hadoop.fs.Path>>> walsByIdRecoveredQueues

syncReplicationPeerMappingManager

private final SyncReplicationPeerMappingManager syncReplicationPeerMappingManager

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

totalBufferUsed

private AtomicLong totalBufferUsed

sleepForRetries

private final long sleepForRetries

maxRetriesMultiplier

private final int maxRetriesMultiplier

totalBufferLimit

private final long totalBufferLimit

globalMetrics

private final MetricsReplicationGlobalSourceSource globalMetrics

Constructor Detail

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,
                                SyncReplicationPeerMappingManager syncReplicationPeerMappingManager,
                                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 Detail

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(ReplicationQueueData queueData,
                                                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

void addSource(String peerId,
               boolean init)
        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.

We add a init parameter to indicate whether this is part of the initialization process. If so, we should skip adding the replication queues as this may introduce dead lock on region server start up and hbase:replication table online.

Parameters:

peerId - the id of the replication peer

init - whether this call is part of the initialization process

Throws:

IOException

drainSources

public void drainSources(String peerId)
                  throws IOException,
                         ReplicationException

This is used when we transit a sync replication peer to SyncReplicationState.STANDBY.

When transiting to SyncReplicationState.STANDBY, we can remove all the pending wal files for a replication peer as we do not need to replicate them any more. And this is necessary, otherwise when we transit back to SyncReplicationState.DOWNGRADE_ACTIVE later, the stale data will be replicated again and cause inconsistency.

See HBASE-20426 for more details.

Parameters:

peerId - the id of the sync replication peer

Throws:

IOException

ReplicationException

createRefreshedSource

private ReplicationSourceInterface createRefreshedSource(ReplicationQueueId queueId,
                                                         ReplicationPeer peer)
                                                  throws IOException,
                                                         ReplicationException

Throws:

IOException

ReplicationException

refreshSources

public void refreshSources(String peerId)
                    throws ReplicationException,
                           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:

ReplicationException

IOException

removeRecoveredSource

private boolean removeRecoveredSource(ReplicationSourceInterface src)

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

Parameters:

src - source to clear

finishRecoveredSource

void finishRecoveredSource(ReplicationSourceInterface src)

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(ReplicationQueueId 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:

source - the replication source

entryBatch - the wal entry batch we just shipped

cleanOldLogs

void cleanOldLogs(String log,
                  boolean inclusive,
                  ReplicationSourceInterface source)

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

source - the replication source

removeRemoteWALs

private void removeRemoteWALs(String peerId,
                              String remoteWALDir,
                              Collection<String> wals)
                       throws IOException

Throws:

IOException

cleanOldLogs

private void cleanOldLogs(NavigableSet<String> wals,
                          ReplicationSourceInterface source)

postLogRoll

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

Throws:

IOException

shouldReplicate

private boolean shouldReplicate(ReplicationGroupOffset offset,
                                String wal)

Check whether we should replicate the given wal.

Parameters:

wal - the file name of the wal

Returns:

true means we should replicate the given wal, otherwise false.

claimQueue

void claimQueue(ReplicationQueueId queueId)

getWALFilesToReplicate

private PriorityQueue<org.apache.hadoop.fs.Path> getWALFilesToReplicate(ServerName sourceRS,
                                                                        boolean syncUp,
                                                                        Map<String,ReplicationGroupOffset> offsets)
                                                                 throws IOException

Throws:

IOException

addRecoveredSource

private void addRecoveredSource(ReplicationSourceInterface src,
                                ReplicationPeerImpl oldPeer,
                                ReplicationQueueId claimedQueueId,
                                PriorityQueue<org.apache.hadoop.fs.Path> walFiles)

claimQueue

void claimQueue(ReplicationQueueId queueId,
                boolean syncUp)

Claim a replication queue.

We add a flag to indicate whether we are called by ReplicationSyncUp. For normal claiming queue operation, we are the last step of a SCP, so we can assume that all the WAL files are under oldWALs directory. But for ReplicationSyncUp, we may want to claim the replication queue for a region server which has not been processed by SCP yet, so we still need to look at its WALs directory.

Parameters:

queueId - the replication queue id we want to claim

syncUp - whether we are called by ReplicationSyncUp

join

public void join()

Terminate the replication on this region server

getWALs

public Map<ReplicationQueueId,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.

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

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.