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

Class ReplicationSource

java.lang.Object

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

All Implemented Interfaces:

ReplicationSourceInterface

Direct Known Subclasses:

RecoveredReplicationSource

@Private
public class ReplicationSource
extends Object
implements ReplicationSourceInterface

Class that handles the source of a replication stream. Currently does not handle more than 1 slave cluster. 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.

Field Summary

Fields

Modifier and Type	Field	Description
private boolean	abortOnError
private final List<WALEntryFilter>	baseFilterOutWALEntries	Base WALEntry filters for this class.
private UUID	clusterId
protected org.apache.hadoop.conf.Configuration	conf
private long	currentBandwidth
static final int	DEFAULT_WAIT_ON_ENDPOINT_SECONDS
private long	defaultBandwidth
private final Predicate<org.apache.hadoop.fs.Path>	filterInWALs	WALs to replicate.
protected org.apache.hadoop.fs.FileSystem	fs
private Thread	initThread
private static final org.slf4j.Logger	LOG
protected ReplicationSourceLogQueue	logQueue
protected ReplicationSourceManager	manager
private int	maxRetriesMultiplier
private MetricsSource	metrics
protected ReplicationQueueId	queueId
protected int	queueSizePerGroup
protected ReplicationQueueStorage	queueStorage
private ReplicationEndpoint	replicationEndpoint
protected ReplicationPeer	replicationPeer
private AtomicBoolean	retryStartup
protected Server	server
private long	sleepForRetries
(package private) boolean	sourceRunning
protected org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap<String,ReplicationGroupOffset>	startOffsets
private AtomicBoolean	startupOngoing
private ReplicationThrottler	throttler
private AtomicLong	totalReplicatedEdits
static final String	WAIT_ON_ENDPOINT_SECONDS
private int	waitOnEndpointSeconds
protected WALEntryFilter	walEntryFilter	A filter (or a chain of filters) for WAL entries; filters out edits.
private WALFileLengthProvider	walFileLengthProvider
protected final ConcurrentHashMap<String,ReplicationSourceShipper>	workerThreads

Constructor Summary

Constructors

Constructor	Description
ReplicationSource()
ReplicationSource(Predicate<org.apache.hadoop.fs.Path> replicateWAL, List<WALEntryFilter> baseFilterOutWALEntries)

Method Summary

Modifier and Type	Method	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	checkError(Thread t, Throwable error)
protected final ReplicationSourceShipper	createNewShipper(String walGroupId)
protected ReplicationSourceShipper	createNewShipper(String walGroupId, ReplicationSourceWALReader walReader)
private ReplicationSourceWALReader	createNewWALReader(String walGroupId, long startPosition)
private ReplicationEndpoint	createReplicationEndpoint()
private void	decorateConf()
void	enqueueLog(org.apache.hadoop.fs.Path wal)	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
private long	getFileSize(org.apache.hadoop.fs.Path currentPath)
ReplicationPeer	getPeer()	Get the replication peer instance.
ReplicationQueueId	getQueueId()	Get the queue id that the source is replicating to
Map<String,PriorityBlockingQueue<org.apache.hadoop.fs.Path>>	getQueues()
ReplicationEndpoint	getReplicationEndpoint()	Returns the replication endpoint used by this replication source
ReplicationQueueStorage	getReplicationQueueStorage()	Returns The instance of queueStorage used by this ReplicationSource.
(package private) Server	getServer()
ServerName	getServerWALsBelongTo()	The queue of WALs only belong to one region server.
ReplicationSourceManager	getSourceManager()	Returns the replication source manager
MetricsSource	getSourceMetrics()	Returns metrics of this replication source
private long	getStartOffset(String walGroupId)
String	getStats()	Get a string representation of the current statistics for this source
long	getTotalReplicatedEdits()
(package private) WALEntryFilter	getWalEntryFilter()	Call after initializeWALEntryFilter(UUID) else it will be null.
WALFileLengthProvider	getWALFileLengthProvider()	Returns the wal file length provider
Map<String,ReplicationStatus>	getWalGroupStatus()	get the stat of replication for each wal group.
void	init(org.apache.hadoop.conf.Configuration conf, org.apache.hadoop.fs.FileSystem fs, ReplicationSourceManager manager, ReplicationQueueStorage queueStorage, ReplicationPeer replicationPeer, Server server, ReplicationQueueData queueData, UUID clusterId, WALFileLengthProvider walFileLengthProvider, MetricsSource metrics)	Instantiation method used by region servers
private void	initAndStartReplicationEndpoint(ReplicationEndpoint replicationEndpoint)
private void	initialize()
private void	initializeWALEntryFilter(UUID peerClusterId)
boolean	isSourceActive()	Returns active or not
boolean	isWorkerRunning()
String	logPeerId()
void	postShipEdits(List<WAL.Entry> entries, long batchSize)	Call this after the shipper thread ship some entries to peer cluster.
(package private) void	removeWorker(ReplicationSourceShipper worker)
private void	retryRefreshing(Thread t, Throwable error)
private void	setSourceStartupStatus(boolean initializing)
private boolean	sleepForRetries(String msg, int sleepMultiplier)	Do the sleeping logic
protected final void	startShipper(ReplicationSourceShipper worker)
protected void	startShippers()
ReplicationSourceInterface	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 clearMetrics)	End the replication
private void	terminate(String reason, Exception cause, boolean clearMetrics, boolean join)
private void	tryStartNewShipper(String walGroupId)
void	tryThrottle(int batchSize)	Try to throttle when the peer config with a bandwidth

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceInterface

getPeerId, isPeerEnabled, isRecovered, isSyncReplication, logPositionAndCleanOldLogs

Field Details

LOG

private static final org.slf4j.Logger LOG

queueSizePerGroup

protected int queueSizePerGroup

logQueue

protected ReplicationSourceLogQueue logQueue

queueStorage

protected ReplicationQueueStorage queueStorage

replicationPeer

protected ReplicationPeer replicationPeer

conf

protected org.apache.hadoop.conf.Configuration conf

manager

protected ReplicationSourceManager manager

server

protected Server server

sleepForRetries

private long sleepForRetries

fs

protected org.apache.hadoop.fs.FileSystem fs

clusterId

private UUID clusterId

totalReplicatedEdits

private AtomicLong totalReplicatedEdits

queueId

protected ReplicationQueueId queueId

startOffsets

protected org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap<String,ReplicationGroupOffset> startOffsets

maxRetriesMultiplier

private int maxRetriesMultiplier

sourceRunning

volatile boolean sourceRunning

metrics

private MetricsSource metrics

replicationEndpoint

private volatile ReplicationEndpoint replicationEndpoint

abortOnError

private boolean abortOnError

startupOngoing

private AtomicBoolean startupOngoing

retryStartup

private AtomicBoolean retryStartup

walEntryFilter

protected volatile WALEntryFilter walEntryFilter

A filter (or a chain of filters) for WAL entries; filters out edits.

throttler

private ReplicationThrottler throttler

defaultBandwidth

private long defaultBandwidth

currentBandwidth

private long currentBandwidth

walFileLengthProvider

private WALFileLengthProvider walFileLengthProvider

workerThreads

protected final ConcurrentHashMap<String,ReplicationSourceShipper> workerThreads

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

initThread

private Thread initThread

filterInWALs

private final Predicate<org.apache.hadoop.fs.Path> filterInWALs

WALs to replicate. Predicate that returns 'true' for WALs to replicate and false for WALs to skip.

baseFilterOutWALEntries

private final List<WALEntryFilter> baseFilterOutWALEntries

Base WALEntry filters for this class. Unmodifiable. Set on construction. Filters *out* edits we do not want replicated, passed on to replication endpoints. This is the basic set. Down in #initializeWALEntryFilter this set is added to the end of the WALEntry filter chain. These are put after those that we pick up from the configured endpoints and other machinations to create the final walEntryFilter.

See Also:

• WALEntryFilter

Constructor Details

ReplicationSource

ReplicationSource()

ReplicationSource

ReplicationSource(Predicate<org.apache.hadoop.fs.Path> replicateWAL,
 List<WALEntryFilter> baseFilterOutWALEntries)

Parameters:

replicateWAL - Pass a filter to run against WAL Path; filter *in* WALs to Replicate; i.e. return 'true' if you want to replicate the content of the WAL.

baseFilterOutWALEntries - Base set of filters you want applied always; filters *out* WALEntries so they never make it out of this ReplicationSource.

Method Details

init

public void init(org.apache.hadoop.conf.Configuration conf,
 org.apache.hadoop.fs.FileSystem fs,
 ReplicationSourceManager manager,
 ReplicationQueueStorage queueStorage,
 ReplicationPeer replicationPeer,
 Server server,
 ReplicationQueueData queueData,
 UUID clusterId,
 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

queueData - the id and offsets of our replication queue

clusterId - unique UUID for the cluster

metrics - metrics for replication source

queueStorage - the replication queue storage

replicationPeer - the replication peer

walFileLengthProvider - for getting the length of the WAL file which is currently being written

Throws:

IOException

decorateConf

private void decorateConf()

enqueueLog

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

Description copied from interface: ReplicationSourceInterface

Add a log to the list of logs to replicate

Specified by:

enqueueLog in interface ReplicationSourceInterface

Parameters:

wal - path to the log to replicate

getQueues

public Map<String,PriorityBlockingQueue<org.apache.hadoop.fs.Path>> getQueues()

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

createReplicationEndpoint

private ReplicationEndpoint createReplicationEndpoint()
                                               throws InstantiationException,
IllegalAccessException,
ClassNotFoundException,
IOException

Throws:

InstantiationException

IllegalAccessException

ClassNotFoundException

IOException

initAndStartReplicationEndpoint

private void initAndStartReplicationEndpoint(ReplicationEndpoint replicationEndpoint)
                                      throws IOException,
TimeoutException

Throws:

IOException

TimeoutException

initializeWALEntryFilter

private void initializeWALEntryFilter(UUID peerClusterId)

getStartOffset

private long getStartOffset(String walGroupId)

createNewShipper

protected final ReplicationSourceShipper createNewShipper(String walGroupId)

startShipper

protected final void startShipper(ReplicationSourceShipper worker)

tryStartNewShipper

private void tryStartNewShipper(String walGroupId)

getWalGroupStatus

public Map<String,ReplicationStatus> getWalGroupStatus()

Description copied from interface: ReplicationSourceInterface

get the stat of replication for each wal group.

Specified by:

getWalGroupStatus in interface ReplicationSourceInterface

Returns:

stat of replication

getFileSize

private long getFileSize(org.apache.hadoop.fs.Path currentPath)
                  throws IOException

Throws:

IOException

createNewShipper

protected ReplicationSourceShipper createNewShipper(String walGroupId,
 ReplicationSourceWALReader walReader)

createNewWALReader

private ReplicationSourceWALReader createNewWALReader(String walGroupId,
 long startPosition)

getWalEntryFilter

WALEntryFilter getWalEntryFilter()

Call after initializeWALEntryFilter(UUID) else it will be null.

Returns:

WAL Entry Filter Chain to use on WAL files filtering *out* WALEntry edits.

checkError

private void checkError(Thread t,
 Throwable error)

retryRefreshing

private void retryRefreshing(Thread t,
 Throwable error)

getReplicationEndpoint

public ReplicationEndpoint getReplicationEndpoint()

Description copied from interface: ReplicationSourceInterface

Returns the replication endpoint used by this replication source

Specified by:

getReplicationEndpoint in interface ReplicationSourceInterface

getSourceManager

public ReplicationSourceManager getSourceManager()

Description copied from interface: ReplicationSourceInterface

Returns the replication source manager

Specified by:

getSourceManager in interface ReplicationSourceInterface

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

sleepForRetries

private 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

initialize

private void initialize()

startShippers

protected void startShippers()

setSourceStartupStatus

private void setSourceStartupStatus(boolean initializing)

startup

public ReplicationSourceInterface 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 clearMetrics)

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

clearMetrics - removes all metrics about this Source

terminate

private void terminate(String reason,
 Exception cause,
 boolean clearMetrics,
 boolean join)

getQueueId

public ReplicationQueueId getQueueId()

Description copied from interface: ReplicationSourceInterface

Get the queue id that the source is replicating to

Specified by:

getQueueId in interface ReplicationSourceInterface

Returns:

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

Description copied from interface: ReplicationSourceInterface

Returns active or not

Specified by:

isSourceActive in interface ReplicationSourceInterface

isWorkerRunning

public boolean isWorkerRunning()

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

Description copied from interface: ReplicationSourceInterface

Returns metrics of this replication source

Specified by:

getSourceMetrics in interface ReplicationSourceInterface

postShipEdits

public void postShipEdits(List<WAL.Entry> entries,
 long 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()

Description copied from interface: ReplicationSourceInterface

Returns the wal file length provider

Specified by:

getWALFileLengthProvider in interface ReplicationSourceInterface

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

getPeer

public ReplicationPeer getPeer()

Description copied from interface: ReplicationSourceInterface

Get the replication peer instance.

Specified by:

getPeer in interface ReplicationSourceInterface

Returns:

the replication peer instance

getServer

Server getServer()

getReplicationQueueStorage

public ReplicationQueueStorage getReplicationQueueStorage()

Description copied from interface: ReplicationSourceInterface

Returns The instance of queueStorage used by this ReplicationSource.

Specified by:

getReplicationQueueStorage in interface ReplicationSourceInterface

removeWorker

void removeWorker(ReplicationSourceShipper worker)

logPeerId

public String logPeerId()

getTotalReplicatedEdits

public long getTotalReplicatedEdits()