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 up in ZK

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. This class implements the ReplicationListener interface so that it can track changes in replication state.

Nested Class Summary

Nested Classes

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

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 boolean	replicationForBulkLoadDataEnabled
private ReplicationPeers	replicationPeers
private ReplicationQueues	replicationQueues
private ReplicationTracker	replicationTracker
private Server	server
private long	sleepBeforeFailover
private List<ReplicationSourceInterface>	sources
private AtomicLong	totalBufferUsed
private WALFileLengthProvider	walFileLengthProvider
private Map<String,Map<String,SortedSet<String>>>	walsById
private Map<String,Map<String,SortedSet<String>>>	walsByIdRecoveredQueues

Constructor Summary

Constructors

Constructor and Description

ReplicationSourceManager(ReplicationQueues replicationQueues, 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)
(package private) ReplicationSourceInterface	addSource(String id) Add sources for the given peer cluster on this region server.
private void	cleanOldLogs(SortedSet<String> wals, String key, String id)
void	cleanOldLogs(String key, String id, boolean queueRecovered) Cleans a log file and all older files from ZK.
void	cleanUpHFileRefs(String peerId, List<String> files)
void	closeQueue(ReplicationSourceInterface src) Clear the references to the specified old source
void	closeRecoveredQueue(ReplicationSourceInterface src) Clear the references to the specified old source
void	deleteSource(String peerId, boolean closeConnection) Delete a complete queue of wals associated with a peer cluster
(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 old sources of this rs
ReplicationPeers	getReplicationPeers() Get the ReplicationPeers used by this ReplicationSourceManager
private ReplicationSourceInterface	getReplicationSource(org.apache.hadoop.conf.Configuration conf, org.apache.hadoop.fs.FileSystem fs, ReplicationSourceManager manager, ReplicationQueues replicationQueues, ReplicationPeers replicationPeers, Server server, String peerId, UUID clusterId, ReplicationPeerConfig peerConfig, ReplicationPeer replicationPeer, WALFileLengthProvider walFileLengthProvider) Factory method to create a replication source
(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()
(package private) Map<String,Map<String,SortedSet<String>>>	getWALs() Get a copy of the wals of the first source on this rs
(package private) Map<String,Map<String,SortedSet<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 and tries to process all old region server wal queues
void	join() Terminate the replication on this region server
void	logPositionAndCleanOldLogs(org.apache.hadoop.fs.Path log, String id, long position, boolean queueRecovered, boolean holdLogInZK) Provide the id of the peer and a log key and this method will figure which wal it belongs to and will log, for this region server, the current position.
void	peerListChanged(List<String> peerIds) The list of registered peer clusters has changed.
void	peerRemoved(String peerId) A peer cluster has been removed (i.e.
void	postLogRoll(org.apache.hadoop.fs.Path newLog)
void	preLogRoll(org.apache.hadoop.fs.Path newLog)
private void	recordLog(org.apache.hadoop.fs.Path logPath) Check and enqueue the given log to the correct source.
void	regionServerRemoved(String regionserver) A region server has been removed from the local cluster
void	removePeer(String id) Thie method first deletes all the recovered sources for the specified id, then deletes the normal source (deleting all related data in ZK).
private void	transferQueues(String rsZnode) 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 List<ReplicationSourceInterface> sources

oldsources

private final List<ReplicationSourceInterface> oldsources

replicationQueues

private final ReplicationQueues replicationQueues

replicationTracker

private final ReplicationTracker replicationTracker

replicationPeers

private final ReplicationPeers replicationPeers

clusterId

private final UUID clusterId

server

private final Server server

walsById

private final Map<String,Map<String,SortedSet<String>>> walsById

walsByIdRecoveredQueues

private final Map<String,Map<String,SortedSet<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(ReplicationQueues replicationQueues,
                                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:

replicationQueues - 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

logPositionAndCleanOldLogs

public void logPositionAndCleanOldLogs(org.apache.hadoop.fs.Path log,
                                       String id,
                                       long position,
                                       boolean queueRecovered,
                                       boolean holdLogInZK)

Provide the id of the peer and a log key and this method will figure which wal it belongs to and will log, for this region server, the current position. It will also clean old logs from the queue.

Parameters:

log - Path to the log currently being replicated from replication status in zookeeper. It will also delete older entries.

id - id of the peer cluster

position - current location in the log

queueRecovered - indicates if this queue comes from another region server

holdLogInZK - if true then the log is retained in ZK

cleanOldLogs

public void cleanOldLogs(String key,
                         String id,
                         boolean queueRecovered)

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

Parameters:

key - Path to the log

id - id of the peer cluster

queueRecovered - Whether this is a recovered queue

cleanOldLogs

private void cleanOldLogs(SortedSet<String> wals,
                          String key,
                          String id)

init

void init()
   throws IOException,
          ReplicationException

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

Throws:

IOException

ReplicationException

addSource

ReplicationSourceInterface addSource(String id)
                              throws IOException,
                                     ReplicationException

Add sources for the given peer cluster on this region server. For the newly added peer, we only need to enqueue the latest log of each wal group and do replication

Parameters:

id - the id of the peer cluster

Returns:

the source that was created

Throws:

IOException

ReplicationException

getSizeOfLatestPath

int getSizeOfLatestPath()

deleteSource

public void deleteSource(String peerId,
                         boolean closeConnection)

Delete a complete queue of wals associated with a peer cluster

Parameters:

peerId - Id of the peer cluster queue of wals to delete

join

public void join()

Terminate the replication on this region server

getWALs

Map<String,Map<String,SortedSet<String>>> getWALs()

Get a copy of the wals of the first source on this rs

Returns:

a sorted set of wal names

getWalsByIdRecoveredQueues

Map<String,Map<String,SortedSet<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:

lis of all sources

getOldSources

public List<ReplicationSourceInterface> getOldSources()

Get a list of all the old sources of this rs

Returns:

list of all old sources

getSource

public ReplicationSourceInterface getSource(String peerId)

Get the normal source for a given peer

Parameters:

peerId -

Returns:

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

getAllQueues

List<String> getAllQueues()

preLogRoll

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

Throws:

IOException

recordLog

private void recordLog(org.apache.hadoop.fs.Path logPath)
                throws IOException

Check and enqueue the given log to the correct source. If there's still no source for the group to which the given log belongs, create one

Parameters:

logPath - the log path to check and enqueue

Throws:

IOException

postLogRoll

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

Throws:

IOException

getTotalBufferUsed

public AtomicLong getTotalBufferUsed()

getReplicationSource

private ReplicationSourceInterface getReplicationSource(org.apache.hadoop.conf.Configuration conf,
                                                        org.apache.hadoop.fs.FileSystem fs,
                                                        ReplicationSourceManager manager,
                                                        ReplicationQueues replicationQueues,
                                                        ReplicationPeers replicationPeers,
                                                        Server server,
                                                        String peerId,
                                                        UUID clusterId,
                                                        ReplicationPeerConfig peerConfig,
                                                        ReplicationPeer replicationPeer,
                                                        WALFileLengthProvider walFileLengthProvider)
                                                 throws IOException

Factory method to create a replication source

Parameters:

conf - the configuration to use

fs - the file system to use

manager - the manager to use

server - the server object for this region server

peerId - the id of the peer cluster

Returns:

the created source

Throws:

IOException

transferQueues

private void transferQueues(String rsZnode)

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 znodes and finally will delete the old znodes. It creates one old source for any type of source of the old rs.

Parameters:

rsZnode -

closeRecoveredQueue

public void closeRecoveredQueue(ReplicationSourceInterface src)

Clear the references to the specified old source

Parameters:

src - source to clear

closeQueue

public void closeQueue(ReplicationSourceInterface src)

Clear the references to the specified old source

Parameters:

src - source to clear

removePeer

public void removePeer(String id)

Thie method first deletes all the recovered sources for the specified id, then deletes the normal source (deleting all related data in ZK).

Parameters:

id - The id of the peer cluster

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

peerRemoved

public void peerRemoved(String peerId)

Description copied from interface: ReplicationListener

A peer cluster has been removed (i.e. unregistered) from replication.

Specified by:

peerRemoved in interface ReplicationListener

Parameters:

peerId - The peer id of the cluster that has been removed

peerListChanged

public void peerListChanged(List<String> peerIds)

Description copied from interface: ReplicationListener

The list of registered peer clusters has changed.

Specified by:

peerListChanged in interface ReplicationListener

Parameters:

peerIds - A list of all currently registered peer clusters

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 ReplicationException

Throws:

ReplicationException

cleanUpHFileRefs

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