org.apache.hadoop.hbase.replication.regionserver

Class ReplicationSource

java.lang.Object

java.lang.Thread

org.apache.hadoop.hbase.replication.regionserver.ReplicationSource

All Implemented Interfaces:

Runnable, ReplicationSourceInterface

Direct Known Subclasses:

RecoveredReplicationSource

@InterfaceAudience.Private
public class ReplicationSource
extends Thread
implements ReplicationSourceInterface

Class that handles the source of a replication stream. Currently does not handle more than 1 slave For each slave cluster it selects a random number of peers using a replication ratio. For example, if replication ration = 0.1 and slave cluster has 100 region servers, 10 will be selected.

A stream is considered down when we cannot contact a region server on the peer cluster for more than 55 seconds by default.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	ReplicationSource.LogsComparator Comparator used to compare logs together based on their start time

Nested classes/interfaces inherited from class java.lang.Thread

Thread.State, Thread.UncaughtExceptionHandler

Field Summary

Fields

Modifier and Type	Field and Description
private UUID	clusterId
protected org.apache.hadoop.conf.Configuration	conf
private long	currentBandwidth
static int	DEFAULT_WAIT_ON_ENDPOINT_SECONDS
private long	defaultBandwidth
protected org.apache.hadoop.fs.FileSystem	fs
private static org.slf4j.Logger	LOG
private int	logQueueWarnThreshold
protected ReplicationSourceManager	manager
private int	maxRetriesMultiplier
private MetricsSource	metrics
private UUID	peerClusterId
protected String	peerClusterZnode
private String	peerId
private Map<String,PriorityBlockingQueue<org.apache.hadoop.fs.Path>>	queues
protected int	queueSizePerGroup
private ReplicationEndpoint	replicationEndpoint
private ReplicationPeers	replicationPeers
protected ReplicationQueueInfo	replicationQueueInfo
protected ReplicationQueues	replicationQueues
protected Server	server
private long	sleepForRetries
private boolean	sourceRunning
private ReplicationThrottler	throttler
private AtomicLong	totalBufferUsed
private AtomicLong	totalReplicatedEdits
static String	WAIT_ON_ENDPOINT_SECONDS
private int	waitOnEndpointSeconds
protected WALEntryFilter	walEntryFilter
private WALFileLengthProvider	walFileLengthProvider
protected ConcurrentHashMap<String,ReplicationSourceShipper>	workerThreads

Fields inherited from class java.lang.Thread

MAX_PRIORITY, MIN_PRIORITY, NORM_PRIORITY

Constructor Summary

Constructors

Constructor and Description
ReplicationSource()

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) Add hfile names to the queue to be replicated.
private void	checkBandwidthChangeAndResetThrottler()
private void	decorateConf()
void	enqueueLog(org.apache.hadoop.fs.Path log) Add a log to the list of logs to replicate
private long	getCurrentBandwidth()
org.apache.hadoop.fs.Path	getCurrentPath() Get the current log that's replicated
String	getPeerClusterZnode() Get the id that the source is replicating to
String	getPeerId() Get the id that the source is replicating to.
ReplicationEndpoint	getReplicationEndpoint()
ServerName	getServerWALsBelongTo() The queue of WALs only belong to one region server.
ReplicationSourceManager	getSourceManager()
MetricsSource	getSourceMetrics()
String	getStats() Get a string representation of the current statistics for this source
Thread.UncaughtExceptionHandler	getUncaughtExceptionHandler()
WALFileLengthProvider	getWALFileLengthProvider()
void	init(org.apache.hadoop.conf.Configuration conf, org.apache.hadoop.fs.FileSystem fs, ReplicationSourceManager manager, ReplicationQueues replicationQueues, ReplicationPeers replicationPeers, Server server, String peerClusterZnode, UUID clusterId, ReplicationEndpoint replicationEndpoint, WALFileLengthProvider walFileLengthProvider, MetricsSource metrics) Instantiation method used by region servers
private void	initializeWALEntryFilter()
boolean	isPeerEnabled() check whether the peer is enabled or not
boolean	isSourceActive()
void	postShipEdits(List<WAL.Entry> entries, int batchSize) Call this after the shipper thread ship some entries to peer cluster.
void	run()
protected boolean	sleepForRetries(String msg, int sleepMultiplier) Do the sleeping logic
protected ReplicationSourceWALReader	startNewWALReader(String threadName, String walGroupId, PriorityBlockingQueue<org.apache.hadoop.fs.Path> queue, long startPosition)
void	startup() Start the replication
void	terminate(String reason) End the replication
void	terminate(String reason, Exception cause) End the replication
void	terminate(String reason, Exception cause, boolean join)
protected void	tryStartNewShipper(String walGroupId, PriorityBlockingQueue<org.apache.hadoop.fs.Path> queue)
void	tryThrottle(int batchSize) Try to throttle when the peer config with a bandwidth
private void	uninitialize()

Methods inherited from class java.lang.Thread

activeCount, checkAccess, clone, countStackFrames, currentThread, destroy, dumpStack, enumerate, getAllStackTraces, getContextClassLoader, getDefaultUncaughtExceptionHandler, getId, getName, getPriority, getStackTrace, getState, getThreadGroup, holdsLock, interrupt, interrupted, isAlive, isDaemon, isInterrupted, join, join, join, resume, setContextClassLoader, setDaemon, setDefaultUncaughtExceptionHandler, setName, setPriority, setUncaughtExceptionHandler, sleep, sleep, start, stop, stop, suspend, toString, yield

Methods inherited from class java.lang.Object

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

Field Detail

LOG

private static final org.slf4j.Logger LOG

queues

private Map<String,PriorityBlockingQueue<org.apache.hadoop.fs.Path>> queues

queueSizePerGroup

protected int queueSizePerGroup

replicationQueues

protected ReplicationQueues replicationQueues

replicationPeers

private ReplicationPeers replicationPeers

conf

protected org.apache.hadoop.conf.Configuration conf

replicationQueueInfo

protected ReplicationQueueInfo replicationQueueInfo

peerId

private String peerId

manager

protected ReplicationSourceManager manager

server

protected Server server

sleepForRetries

private long sleepForRetries

fs

protected org.apache.hadoop.fs.FileSystem fs

clusterId

private UUID clusterId

peerClusterId

private UUID peerClusterId

totalReplicatedEdits

private AtomicLong totalReplicatedEdits

peerClusterZnode

protected String peerClusterZnode

maxRetriesMultiplier

private int maxRetriesMultiplier

sourceRunning

private volatile boolean sourceRunning

metrics

private MetricsSource metrics

logQueueWarnThreshold

private int logQueueWarnThreshold

replicationEndpoint

private ReplicationEndpoint replicationEndpoint

walEntryFilter

protected WALEntryFilter walEntryFilter

throttler

private ReplicationThrottler throttler

defaultBandwidth

private long defaultBandwidth

currentBandwidth

private long currentBandwidth

walFileLengthProvider

private WALFileLengthProvider walFileLengthProvider

workerThreads

protected final ConcurrentHashMap<String,ReplicationSourceShipper> workerThreads

totalBufferUsed

private AtomicLong totalBufferUsed

WAIT_ON_ENDPOINT_SECONDS

public static final String WAIT_ON_ENDPOINT_SECONDS

See Also:

Constant Field Values

DEFAULT_WAIT_ON_ENDPOINT_SECONDS

public static final int DEFAULT_WAIT_ON_ENDPOINT_SECONDS

See Also:

Constant Field Values

waitOnEndpointSeconds

private int waitOnEndpointSeconds

Constructor Detail

ReplicationSource

public ReplicationSource()

Method Detail

init

public void init(org.apache.hadoop.conf.Configuration conf,
                 org.apache.hadoop.fs.FileSystem fs,
                 ReplicationSourceManager manager,
                 ReplicationQueues replicationQueues,
                 ReplicationPeers replicationPeers,
                 Server server,
                 String peerClusterZnode,
                 UUID clusterId,
                 ReplicationEndpoint replicationEndpoint,
                 WALFileLengthProvider walFileLengthProvider,
                 MetricsSource metrics)
          throws IOException

Instantiation method used by region servers

Specified by:

init in interface ReplicationSourceInterface

Parameters:

conf - configuration to use

fs - file system to use

manager - replication manager to ping to

server - the server for this region server

peerClusterZnode - the name of our znode

clusterId - unique UUID for the cluster

replicationEndpoint - the replication endpoint implementation

metrics - metrics for replication source

Throws:

IOException

decorateConf

private void decorateConf()

enqueueLog

public void enqueueLog(org.apache.hadoop.fs.Path log)

Description copied from interface: ReplicationSourceInterface

Add a log to the list of logs to replicate

Specified by:

enqueueLog in interface ReplicationSourceInterface

Parameters:

log - path to the log to replicate

addHFileRefs

public void addHFileRefs(TableName tableName,
                         byte[] family,
                         List<Pair<org.apache.hadoop.fs.Path,org.apache.hadoop.fs.Path>> pairs)
                  throws ReplicationException

Description copied from interface: ReplicationSourceInterface

Add hfile names to the queue to be replicated.

Specified by:

addHFileRefs in interface ReplicationSourceInterface

Parameters:

tableName - Name of the table these files belongs to

family - Name of the family these files belong to

pairs - list of pairs of { HFile location in staging dir, HFile path in region dir which will be added in the queue for replication}

Throws:

ReplicationException - If failed to add hfile references

run

public void run()

Specified by:

run in interface Runnable

Overrides:

run in class Thread

initializeWALEntryFilter

private void initializeWALEntryFilter()

tryStartNewShipper

protected void tryStartNewShipper(String walGroupId,
                                  PriorityBlockingQueue<org.apache.hadoop.fs.Path> queue)

startNewWALReader

protected ReplicationSourceWALReader startNewWALReader(String threadName,
                                                       String walGroupId,
                                                       PriorityBlockingQueue<org.apache.hadoop.fs.Path> queue,
                                                       long startPosition)

getUncaughtExceptionHandler

public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler()

Overrides:

getUncaughtExceptionHandler in class Thread

getReplicationEndpoint

public ReplicationEndpoint getReplicationEndpoint()

Specified by:

getReplicationEndpoint in interface ReplicationSourceInterface

Returns:

the replication endpoint used by this replication source

getSourceManager

public ReplicationSourceManager getSourceManager()

Specified by:

getSourceManager in interface ReplicationSourceInterface

Returns:

the replication source manager

tryThrottle

public void tryThrottle(int batchSize)
                 throws InterruptedException

Description copied from interface: ReplicationSourceInterface

Try to throttle when the peer config with a bandwidth

Specified by:

tryThrottle in interface ReplicationSourceInterface

Parameters:

batchSize - entries size will be pushed

Throws:

InterruptedException

checkBandwidthChangeAndResetThrottler

private void checkBandwidthChangeAndResetThrottler()

getCurrentBandwidth

private long getCurrentBandwidth()

uninitialize

private void uninitialize()

sleepForRetries

protected boolean sleepForRetries(String msg,
                                  int sleepMultiplier)

Do the sleeping logic

Parameters:

msg - Why we sleep

sleepMultiplier - by how many times the default sleeping time is augmented

Returns:

True if sleepMultiplier is < maxRetriesMultiplier

isPeerEnabled

public boolean isPeerEnabled()

check whether the peer is enabled or not

Specified by:

isPeerEnabled in interface ReplicationSourceInterface

Returns:

true if the peer is enabled, otherwise false

startup

public void startup()

Description copied from interface: ReplicationSourceInterface

Start the replication

Specified by:

startup in interface ReplicationSourceInterface

terminate

public void terminate(String reason)

Description copied from interface: ReplicationSourceInterface

End the replication

Specified by:

terminate in interface ReplicationSourceInterface

Parameters:

reason - why it's terminating

terminate

public void terminate(String reason,
                      Exception cause)

Description copied from interface: ReplicationSourceInterface

End the replication

Specified by:

terminate in interface ReplicationSourceInterface

Parameters:

reason - why it's terminating

cause - the error that's causing it

terminate

public void terminate(String reason,
                      Exception cause,
                      boolean join)

getPeerClusterZnode

public String getPeerClusterZnode()

Description copied from interface: ReplicationSourceInterface

Get the id that the source is replicating to

Specified by:

getPeerClusterZnode in interface ReplicationSourceInterface

Returns:

peer cluster id

getPeerId

public String getPeerId()

Description copied from interface: ReplicationSourceInterface

Get the id that the source is replicating to.

Specified by:

getPeerId in interface ReplicationSourceInterface

Returns:

peer id

getCurrentPath

public org.apache.hadoop.fs.Path getCurrentPath()

Description copied from interface: ReplicationSourceInterface

Get the current log that's replicated

Specified by:

getCurrentPath in interface ReplicationSourceInterface

Returns:

the current log

isSourceActive

public boolean isSourceActive()

Specified by:

isSourceActive in interface ReplicationSourceInterface

Returns:

active or not

getStats

public String getStats()

Description copied from interface: ReplicationSourceInterface

Get a string representation of the current statistics for this source

Specified by:

getStats in interface ReplicationSourceInterface

Returns:

printable stats

getSourceMetrics

public MetricsSource getSourceMetrics()

Specified by:

getSourceMetrics in interface ReplicationSourceInterface

Returns:

metrics of this replication source

postShipEdits

public void postShipEdits(List<WAL.Entry> entries,
                          int batchSize)

Description copied from interface: ReplicationSourceInterface

Call this after the shipper thread ship some entries to peer cluster.

Specified by:

postShipEdits in interface ReplicationSourceInterface

Parameters:

entries - pushed

batchSize - entries size pushed

getWALFileLengthProvider

public WALFileLengthProvider getWALFileLengthProvider()

Specified by:

getWALFileLengthProvider in interface ReplicationSourceInterface

Returns:

the wal file length provider

getServerWALsBelongTo

public ServerName getServerWALsBelongTo()

Description copied from interface: ReplicationSourceInterface

The queue of WALs only belong to one region server. This will return the server name which all WALs belong to.

Specified by:

getServerWALsBelongTo in interface ReplicationSourceInterface

Returns:

the server name which all WALs belong to