org.apache.hadoop.hbase.replication.regionserver

Class HBaseInterClusterReplicationEndpoint

java.lang.Object

org.apache.hbase.thirdparty.com.google.common.util.concurrent.AbstractService

org.apache.hadoop.hbase.replication.BaseReplicationEndpoint

org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint

org.apache.hadoop.hbase.replication.regionserver.HBaseInterClusterReplicationEndpoint

All Implemented Interfaces:

Abortable, ReplicationEndpoint, ReplicationPeerConfigListener, org.apache.hbase.thirdparty.com.google.common.util.concurrent.Service

@InterfaceAudience.Private
public class HBaseInterClusterReplicationEndpoint
extends HBaseReplicationEndpoint

A ReplicationEndpoint implementation for replicating to another HBase cluster. For the 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/interfaces inherited from class org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint

HBaseReplicationEndpoint.PeerRegionServerListener

Nested classes/interfaces inherited from interface org.apache.hadoop.hbase.replication.ReplicationEndpoint

ReplicationEndpoint.Context, ReplicationEndpoint.ReplicateContext

Nested classes/interfaces inherited from interface org.apache.hbase.thirdparty.com.google.common.util.concurrent.Service

org.apache.hbase.thirdparty.com.google.common.util.concurrent.Service.Listener, org.apache.hbase.thirdparty.com.google.common.util.concurrent.Service.State

Field Summary

Fields

Modifier and Type	Field and Description
private Abortable	abortable
private org.apache.hadoop.fs.Path	baseNamespaceDir
private org.apache.hadoop.conf.Configuration	conf
private ClusterConnection	conn
private static long	DEFAULT_MAX_TERMINATION_WAIT_MULTIPLIER
private boolean	dropOnDeletedColumnFamilies
private boolean	dropOnDeletedTables
private ThreadPoolExecutor	exec
private org.apache.hadoop.fs.Path	hfileArchiveDir
private boolean	isSerial
private long	lastSinkFetchTime
private org.apache.hadoop.conf.Configuration	localConf
private static org.slf4j.Logger	LOG
private int	maxRetriesMultiplier
private long	maxTerminationWait
private int	maxThreads
private MetricsSource	metrics
private boolean	peersSelected
static String	REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY Drop edits for CFs that been deleted from the replication source and target
static String	REPLICATION_DROP_ON_DELETED_TABLE_KEY Drop edits for tables that been deleted from the replication source and target
private boolean	replicationBulkLoadDataEnabled
private String	replicationClusterId
private int	replicationRpcLimit
private ReplicationSinkManager	replicationSinkMgr
private long	sleepForRetries
private int	socketTimeoutMultiplier

Fields inherited from class org.apache.hadoop.hbase.replication.BaseReplicationEndpoint

ctx, REPLICATION_WALENTRYFILTER_CONFIG_KEY

Constructor Summary

Constructors

Constructor and Description
HBaseInterClusterReplicationEndpoint()

Method Summary

Modifier and Type	Method and Description
private void	connectToPeers()
private List<List<WAL.Entry>>	createBatches(List<WAL.Entry> entries) Divide the entries into multiple batches, so that we can replicate each batch in a thread pool concurrently.
protected Connection	createConnection(org.apache.hadoop.conf.Configuration conf)
private List<List<WAL.Entry>>	createParallelBatches(List<WAL.Entry> entries)
protected ReplicationSinkManager	createReplicationSinkManager(Connection conn)
protected Callable<Integer>	createReplicator(List<WAL.Entry> entries, int batchIndex, int timeout)
private List<List<WAL.Entry>>	createSerialBatches(List<WAL.Entry> entries)
private void	decorateConf()
protected void	doStop()
(package private) List<List<WAL.Entry>>	filterNotExistColumnFamilyEdits(List<List<WAL.Entry>> oldEntryList)
(package private) List<List<WAL.Entry>>	filterNotExistTableEdits(List<List<WAL.Entry>> oldEntryList)
private int	getEstimatedEntrySize(WAL.Entry e)
void	init(ReplicationEndpoint.Context context) Initialize the replication endpoint with the given context.
static boolean	isNoSuchColumnFamilyException(Throwable io) Check if there's an NoSuchColumnFamilyException in the caused by stacktrace.
protected boolean	isPeerEnabled()
static boolean	isTableNotFoundException(Throwable io) Check if there's an TableNotFoundException in the caused by stacktrace.
private String	logPeerId()
private long	parallelReplicate(CompletionService<Integer> pool, ReplicationEndpoint.ReplicateContext replicateContext, List<List<WAL.Entry>> batches)
private void	reconnectToPeerCluster()
boolean	replicate(ReplicationEndpoint.ReplicateContext replicateContext) Do the shipping logic
protected int	replicateEntries(List<WAL.Entry> entries, int batchIndex, int timeout)
private int	serialReplicateRegionEntries(List<WAL.Entry> entries, int batchIndex, int timeout)
private boolean	sleepForRetries(String msg, int sleepMultiplier) Do the sleeping logic

Methods inherited from class org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint

abort, disconnect, doStart, fetchSlavesAddresses, getLastRegionServerUpdate, getPeerUUID, getRegionServers, getZkw, isAborted, reconnect, setRegionServers, start, stop

Methods inherited from class org.apache.hadoop.hbase.replication.BaseReplicationEndpoint

canReplicateToSameCluster, getNamespaceTableCfWALEntryFilter, getScopeWALEntryFilter, getWALEntryfilter, isStarting, peerConfigUpdated

Methods inherited from class org.apache.hbase.thirdparty.com.google.common.util.concurrent.AbstractService

addListener, awaitRunning, awaitRunning, awaitRunning, awaitTerminated, awaitTerminated, awaitTerminated, doCancelStart, failureCause, isRunning, notifyFailed, notifyStarted, notifyStopped, startAsync, state, stopAsync, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.hadoop.hbase.Abortable

abort

Methods inherited from interface org.apache.hadoop.hbase.replication.ReplicationEndpoint

awaitRunning, awaitRunning, awaitTerminated, awaitTerminated, failureCause, isRunning

Field Detail

LOG

private static final org.slf4j.Logger LOG

DEFAULT_MAX_TERMINATION_WAIT_MULTIPLIER

private static final long DEFAULT_MAX_TERMINATION_WAIT_MULTIPLIER

See Also:

Constant Field Values

REPLICATION_DROP_ON_DELETED_TABLE_KEY

public static final String REPLICATION_DROP_ON_DELETED_TABLE_KEY

Drop edits for tables that been deleted from the replication source and target

See Also:

Constant Field Values

REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY

public static final String REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY

Drop edits for CFs that been deleted from the replication source and target

See Also:

Constant Field Values

conn

private ClusterConnection conn

localConf

private org.apache.hadoop.conf.Configuration localConf

conf

private org.apache.hadoop.conf.Configuration conf

sleepForRetries

private long sleepForRetries

maxRetriesMultiplier

private int maxRetriesMultiplier

socketTimeoutMultiplier

private int socketTimeoutMultiplier

maxTerminationWait

private long maxTerminationWait

replicationRpcLimit

private int replicationRpcLimit

metrics

private MetricsSource metrics

replicationSinkMgr

private ReplicationSinkManager replicationSinkMgr

peersSelected

private boolean peersSelected

replicationClusterId

private String replicationClusterId

exec

private ThreadPoolExecutor exec

maxThreads

private int maxThreads

baseNamespaceDir

private org.apache.hadoop.fs.Path baseNamespaceDir

hfileArchiveDir

private org.apache.hadoop.fs.Path hfileArchiveDir

replicationBulkLoadDataEnabled

private boolean replicationBulkLoadDataEnabled

abortable

private Abortable abortable

dropOnDeletedTables

private boolean dropOnDeletedTables

dropOnDeletedColumnFamilies

private boolean dropOnDeletedColumnFamilies

isSerial

private boolean isSerial

lastSinkFetchTime

private long lastSinkFetchTime

Constructor Detail

HBaseInterClusterReplicationEndpoint

public HBaseInterClusterReplicationEndpoint()

Method Detail

createConnection

protected Connection createConnection(org.apache.hadoop.conf.Configuration conf)
                               throws IOException

Throws:

IOException

createReplicationSinkManager

protected ReplicationSinkManager createReplicationSinkManager(Connection conn)

init

public void init(ReplicationEndpoint.Context context)
          throws IOException

Description copied from interface: ReplicationEndpoint

Initialize the replication endpoint with the given context.

Specified by:

init in interface ReplicationEndpoint

Overrides:

init in class BaseReplicationEndpoint

Parameters:

context - replication context

Throws:

IOException - error occur when initialize the endpoint.

decorateConf

private void decorateConf()

connectToPeers

private void connectToPeers()

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

getEstimatedEntrySize

private int getEstimatedEntrySize(WAL.Entry e)

createParallelBatches

private List<List<WAL.Entry>> createParallelBatches(List<WAL.Entry> entries)

createSerialBatches

private List<List<WAL.Entry>> createSerialBatches(List<WAL.Entry> entries)

createBatches

private List<List<WAL.Entry>> createBatches(List<WAL.Entry> entries)

Divide the entries into multiple batches, so that we can replicate each batch in a thread pool concurrently. Note that, for serial replication, we need to make sure that entries from the same region to be replicated serially, so entries from the same region consist of a batch, and we will divide a batch into several batches by replicationRpcLimit in method serialReplicateRegionEntries()

isTableNotFoundException

public static boolean isTableNotFoundException(Throwable io)

Check if there's an TableNotFoundException in the caused by stacktrace.

isNoSuchColumnFamilyException

public static boolean isNoSuchColumnFamilyException(Throwable io)

Check if there's an NoSuchColumnFamilyException in the caused by stacktrace.

filterNotExistTableEdits

List<List<WAL.Entry>> filterNotExistTableEdits(List<List<WAL.Entry>> oldEntryList)

filterNotExistColumnFamilyEdits

List<List<WAL.Entry>> filterNotExistColumnFamilyEdits(List<List<WAL.Entry>> oldEntryList)

reconnectToPeerCluster

private void reconnectToPeerCluster()

parallelReplicate

private long parallelReplicate(CompletionService<Integer> pool,
                               ReplicationEndpoint.ReplicateContext replicateContext,
                               List<List<WAL.Entry>> batches)
                        throws IOException

Throws:

IOException

replicate

public boolean replicate(ReplicationEndpoint.ReplicateContext replicateContext)

Do the shipping logic

Parameters:

replicateContext - a context where WAL entries and other parameters can be obtained.

isPeerEnabled

protected boolean isPeerEnabled()

doStop

protected void doStop()

Overrides:

doStop in class HBaseReplicationEndpoint

replicateEntries

protected int replicateEntries(List<WAL.Entry> entries,
                               int batchIndex,
                               int timeout)
                        throws IOException

Throws:

IOException

serialReplicateRegionEntries

private int serialReplicateRegionEntries(List<WAL.Entry> entries,
                                         int batchIndex,
                                         int timeout)
                                  throws IOException

Throws:

IOException

createReplicator

protected Callable<Integer> createReplicator(List<WAL.Entry> entries,
                                             int batchIndex,
                                             int timeout)

logPeerId

private String logPeerId()