ReplicationSourceManager (Apache HBase 1.2.12 API)

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.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.apache.commons.logging.Log`	`LOG`
`private org.apache.hadoop.fs.Path`	`logDir`
`private org.apache.hadoop.fs.Path`	`oldLogDir`
`private List<ReplicationSourceInterface>`	`oldsources`
`private Random`	`rand`
`private ReplicationPeers`	`replicationPeers`
`private ReplicationQueues`	`replicationQueues`
`private ReplicationTracker`	`replicationTracker`
`private Server`	`server`
`private long`	`sleepBeforeFailover`
`private List<ReplicationSourceInterface>`	`sources`
`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)` Creates a replication manager and sets the watch on all the other registered region servers

Method Summary

Methods
Modifier and Type	Method and Description
`protected 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`	`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
`protected 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)` Factory method to create a replication source
`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
`protected Map<String,Map<String,SortedSet<String>>>`	`getWALs()` Get a copy of the wals of the first source on this rs
`protected Map<String,Map<String,SortedSet<String>>>`	`getWalsByIdRecoveredQueues()` Get a copy of the wals of the recovered sources on this rs
`protected 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.
`(package private) void`	`postLogRoll(org.apache.hadoop.fs.Path newLog)`
`(package private) 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.apache.commons.logging.Log 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 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

sleepBeforeFailover
```
private final long sleepBeforeFailover
```

executor

private final ThreadPoolExecutor executor

rand
```
private final Random rand
```

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)

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 -

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
```
protected 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
```
protected 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
- 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
```
protected 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
```
protected 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
- getAllQueues
```
List<String> getAllQueues()
```
- preLogRoll
```
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
```
void postLogRoll(org.apache.hadoop.fs.Path newLog)
           throws IOException
```
  Throws:
  
  IOException
- getReplicationSource
```
protected 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)
                                                   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
- getStats
```
public String getStats()
```
  Get a string representation of all the sources' metrics

Class ReplicationSourceManager

Nested Class Summary

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

LOG

sources

oldsources

replicationQueues

replicationTracker

replicationPeers

clusterId

server

walsById

walsByIdRecoveredQueues

conf

fs

latestPaths

logDir

oldLogDir

sleepBeforeFailover

executor

rand

Constructor Detail

ReplicationSourceManager

Method Detail

logPositionAndCleanOldLogs

cleanOldLogs

cleanOldLogs

init

addSource

deleteSource

join

getWALs

getWalsByIdRecoveredQueues

getSources

getOldSources

getAllQueues

preLogRoll

recordLog

postLogRoll

getReplicationSource

transferQueues

closeRecoveredQueue

closeQueue

removePeer

regionServerRemoved

peerRemoved

peerListChanged

getOldLogDir

getLogDir

getFs

getStats