Package 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

@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.SinkPeer

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	Description
private org.apache.hadoop.fs.Path	baseNamespaceDir
private boolean	dropOnDeletedColumnFamilies
private boolean	dropOnDeletedTables
private org.apache.hadoop.fs.Path	hfileArchiveDir
private boolean	isSerial
private long	lastSinkFetchTime
private static final org.slf4j.Logger	LOG
private int	maxRetriesMultiplier
private int	maxThreads
private MetricsSource	metrics
private boolean	peersSelected
static final String	REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY	Drop edits for CFs that been deleted from the replication source and target
static final 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 long	sleepForRetries
private int	socketTimeoutMultiplier

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

conf, DEFAULT_BAD_SINK_THRESHOLD, DEFAULT_REPLICATION_SOURCE_RATIO

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

ctx, REPLICATION_WALENTRYFILTER_CONFIG_KEY

Constructor Summary

Constructors

Constructor	Description
HBaseInterClusterReplicationEndpoint()

Method Summary

Modifier and Type	Method	Description
protected CompletableFuture<Integer>	asyncReplicate(List<WAL.Entry> entries, int batchIndex, int timeout)	Replicate entries to peer cluster by async API.
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.
private List<List<WAL.Entry>>	createParallelBatches(List<WAL.Entry> entries)
private List<List<WAL.Entry>>	createSerialBatches(List<WAL.Entry> entries)
private void	decorateConf()
(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 void	onReplicateWALEntryException(int entriesHashCode, Throwable exception, HBaseReplicationEndpoint.SinkPeer sinkPeer)
private long	parallelReplicate(ReplicationEndpoint.ReplicateContext replicateContext, List<List<WAL.Entry>> batches)
boolean	replicate(ReplicationEndpoint.ReplicateContext replicateContext)	Do the shipping logic
protected CompletableFuture<Integer>	replicateEntries(List<WAL.Entry> entries, int batchIndex, int timeout)
private CompletableFuture<Integer>	serialReplicateRegionEntries(org.apache.hbase.thirdparty.com.google.common.collect.PeekingIterator<WAL.Entry> walEntryPeekingIterator, int batchIndex, int timeout)	Here for HBaseInterClusterReplicationEndpoint#isSerialis is true, we iterator over the WAL WAL.Entry list, once we reached a batch limit, we send it out, and in the callback, we send the next batch, until we send all entries out.
private boolean	sleepForRetries(String msg, int sleepMultiplier)	Do the sleeping logic

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

abort, chooseSinks, createConnection, doStart, doStop, fetchPeerAddresses, getNumSinks, getPeerUUID, getReplicationSink, isAborted, reportBadSink, reportSinkSuccess, 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 Details

LOG

private static final org.slf4j.Logger LOG

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

sleepForRetries

private long sleepForRetries

maxRetriesMultiplier

private int maxRetriesMultiplier

socketTimeoutMultiplier

private int socketTimeoutMultiplier

replicationRpcLimit

private int replicationRpcLimit

metrics

private MetricsSource metrics

peersSelected

private boolean peersSelected

replicationClusterId

private String replicationClusterId

maxThreads

private int maxThreads

baseNamespaceDir

private org.apache.hadoop.fs.Path baseNamespaceDir

hfileArchiveDir

private org.apache.hadoop.fs.Path hfileArchiveDir

replicationBulkLoadDataEnabled

private boolean replicationBulkLoadDataEnabled

dropOnDeletedTables

private boolean dropOnDeletedTables

dropOnDeletedColumnFamilies

private boolean dropOnDeletedColumnFamilies

isSerial

private boolean isSerial

lastSinkFetchTime

private long lastSinkFetchTime

Constructor Details

HBaseInterClusterReplicationEndpoint

public HBaseInterClusterReplicationEndpoint()

Method Details

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 HBaseReplicationEndpoint

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)

parallelReplicate

private long parallelReplicate(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()

replicateEntries

protected CompletableFuture<Integer> replicateEntries(List<WAL.Entry> entries,
 int batchIndex,
 int timeout)

onReplicateWALEntryException

private void onReplicateWALEntryException(int entriesHashCode,
 Throwable exception,
 HBaseReplicationEndpoint.SinkPeer sinkPeer)

serialReplicateRegionEntries

private CompletableFuture<Integer> serialReplicateRegionEntries(org.apache.hbase.thirdparty.com.google.common.collect.PeekingIterator<WAL.Entry> walEntryPeekingIterator,
 int batchIndex,
 int timeout)

Here for HBaseInterClusterReplicationEndpoint#isSerialis is true, we iterator over the WAL WAL.Entry list, once we reached a batch limit, we send it out, and in the callback, we send the next batch, until we send all entries out.

asyncReplicate

protected CompletableFuture<Integer> asyncReplicate(List<WAL.Entry> entries,
 int batchIndex,
 int timeout)

Replicate entries to peer cluster by async API.

logPeerId

private String logPeerId()