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

Class ReplicationSink

java.lang.Object

org.apache.hadoop.hbase.replication.regionserver.ReplicationSink

@Private
public class ReplicationSink
extends Object

This class is responsible for replicating the edits coming from another cluster.

This replication process is currently waiting for the edits to be applied before the method can return. This means that the replication of edits is synchronized (after reading from WALs in ReplicationSource) and that a single region server cannot receive edits from two sources at the same time

This class uses the native HBase client in order to replicate entries.

TODO make this class more like ReplicationSource wrt log handling

Field Summary

Fields

Modifier and Type	Field	Description
private final org.apache.hadoop.conf.Configuration	conf
private long	hfilesReplicated
private static final org.slf4j.Logger	LOG
private final MetricsSink	metrics
private SourceFSConfigurationProvider	provider
private boolean	replicationSinkTrackerEnabled
private final int	rowSizeWarnThreshold	Row size threshold for multi requests above which a warning is logged
private final RegionServerCoprocessorHost	rsServerHost
private AsyncConnection	sharedAsyncConnection	This shared AsyncConnection is used for handling wal replication.
private final Object	sharedAsyncConnectionLock
private Connection	sharedConnection	This shared Connection is used for handling bulk load hfiles replication.
private final Object	sharedConnectionLock
private final AtomicLong	totalReplicatedEdits
private WALEntrySinkFilter	walEntrySinkFilter

Constructor Summary

Constructors

Constructor	Description
ReplicationSink(org.apache.hadoop.conf.Configuration conf, RegionServerCoprocessorHost rsServerHost)	Create a sink for replication

Method Summary

Modifier and Type	Method	Description
private void	addFamilyAndItsHFilePathToTableInMap(byte[] family, String pathToHfileFromNS, List<Pair<byte[],List<String>>> familyHFilePathsList)
private void	addNewTableEntryInMap(Map<String,List<Pair<byte[],List<String>>>> bulkLoadHFileMap, byte[] family, String pathToHfileFromNS, String tableName)
private <K1, K2, V> List<V>	addToHashMultiMap(Map<K1,Map<K2,List<V>>> map, K1 key1, K2 key2, V value)	Simple helper to a map from key to (a list of) values TODO: Make a general utility method * * @return the list of values corresponding to key1 and key2
private void	batch(TableName tableName, Collection<List<Row>> allRows, int batchRowSizeThreshold)	Do the changes and handle the pool
private void	buildBulkLoadHFileMap(Map<String,List<Pair<byte[],List<String>>>> bulkLoadHFileMap, TableName table, org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.BulkLoadDescriptor bld)
private void	decorateConf()	decorate the Configuration object to make replication more receptive to delays: lessen the timeout and numTries.
private AsyncConnection	getAsyncConnection()	Return the shared AsyncConnection which is used for handling wal replication.
private Connection	getConnection()	Return the shared Connection which is used for handling bulk load hfiles replication.
private String	getHFilePath(TableName table, org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.BulkLoadDescriptor bld, String storeFile, byte[] family)
MetricsSink	getSinkMetrics()	Get replication Sink Metrics
String	getStats()	Get a string representation of this sink's metrics
private boolean	isNewRowOrType(Cell previousCell, Cell cell)	Returns True if we have crossed over onto a new row or type
private Put	processReplicationMarkerEntry(Cell cell)
void	replicateEntries(List<org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WALEntry> entries, CellScanner cells, String replicationClusterId, String sourceBaseNamespaceDirPath, String sourceHFileArchiveDirPath)	Replicate this array of entries directly into the local cluster using the native client.
private WALEntrySinkFilter	setupWALEntrySinkFilter()
void	stopReplicationSinkServices()	stop the thread pool executor.
private UUID	toUUID(org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.UUID uuid)

Methods inherited from class java.lang.Object

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

Field Details

LOG

private static final org.slf4j.Logger LOG

conf

private final org.apache.hadoop.conf.Configuration conf

sharedConnection

private volatile Connection sharedConnection

This shared Connection is used for handling bulk load hfiles replication.

sharedAsyncConnection

private volatile AsyncConnection sharedAsyncConnection

This shared AsyncConnection is used for handling wal replication.

metrics

private final MetricsSink metrics

totalReplicatedEdits

private final AtomicLong totalReplicatedEdits

sharedConnectionLock

private final Object sharedConnectionLock

sharedAsyncConnectionLock

private final Object sharedAsyncConnectionLock

hfilesReplicated

private long hfilesReplicated

provider

private SourceFSConfigurationProvider provider

walEntrySinkFilter

private WALEntrySinkFilter walEntrySinkFilter

rowSizeWarnThreshold

private final int rowSizeWarnThreshold

Row size threshold for multi requests above which a warning is logged

replicationSinkTrackerEnabled

private boolean replicationSinkTrackerEnabled

rsServerHost

private final RegionServerCoprocessorHost rsServerHost

Constructor Details

ReplicationSink

public ReplicationSink(org.apache.hadoop.conf.Configuration conf,
 RegionServerCoprocessorHost rsServerHost)
                throws IOException

Create a sink for replication

Parameters:

conf - conf object

Throws:

IOException - thrown when HDFS goes bad or bad file name

Method Details

setupWALEntrySinkFilter

private WALEntrySinkFilter setupWALEntrySinkFilter()
                                            throws IOException

Throws:

IOException

decorateConf

private void decorateConf()

decorate the Configuration object to make replication more receptive to delays: lessen the timeout and numTries.

replicateEntries

public void replicateEntries(List<org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WALEntry> entries,
 CellScanner cells,
 String replicationClusterId,
 String sourceBaseNamespaceDirPath,
 String sourceHFileArchiveDirPath)
                      throws IOException

Replicate this array of entries directly into the local cluster using the native client. Only operates against raw protobuf type saving on a conversion from pb to pojo.

Parameters:

entries - WAL entries to be replicated.

cells - cell scanner for iteration.

replicationClusterId - Id which will uniquely identify source cluster FS client configurations in the replication configuration directory

sourceBaseNamespaceDirPath - Path that point to the source cluster base namespace directory

sourceHFileArchiveDirPath - Path that point to the source cluster hfile archive directory

Throws:

IOException - If failed to replicate the data

processReplicationMarkerEntry

private Put processReplicationMarkerEntry(Cell cell)
                                   throws IOException

Throws:

IOException

buildBulkLoadHFileMap

private void buildBulkLoadHFileMap(Map<String,List<Pair<byte[],List<String>>>> bulkLoadHFileMap,
 TableName table,
 org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.BulkLoadDescriptor bld)
                            throws IOException

Throws:

IOException

addFamilyAndItsHFilePathToTableInMap

private void addFamilyAndItsHFilePathToTableInMap(byte[] family,
 String pathToHfileFromNS,
 List<Pair<byte[],List<String>>> familyHFilePathsList)

addNewTableEntryInMap

private void addNewTableEntryInMap(Map<String,List<Pair<byte[],List<String>>>> bulkLoadHFileMap,
 byte[] family,
 String pathToHfileFromNS,
 String tableName)

getHFilePath

private String getHFilePath(TableName table,
 org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.BulkLoadDescriptor bld,
 String storeFile,
 byte[] family)

isNewRowOrType

private boolean isNewRowOrType(Cell previousCell,
 Cell cell)

Returns True if we have crossed over onto a new row or type

toUUID

private UUID toUUID(org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.UUID uuid)

addToHashMultiMap

private <K1,
K2,
V> List<V> addToHashMultiMap(Map<K1,Map<K2,List<V>>> map,
 K1 key1,
 K2 key2,
 V value)

Simple helper to a map from key to (a list of) values TODO: Make a general utility method * * @return the list of values corresponding to key1 and key2

stopReplicationSinkServices

public void stopReplicationSinkServices()

stop the thread pool executor. It is called when the regionserver is stopped.

batch

private void batch(TableName tableName,
 Collection<List<Row>> allRows,
 int batchRowSizeThreshold)
            throws IOException

Do the changes and handle the pool

Parameters:

tableName - table to insert into

allRows - list of actions

batchRowSizeThreshold - rowSize threshold for batch mutation

Throws:

IOException

getConnection

private Connection getConnection()
                          throws IOException

Return the shared Connection which is used for handling bulk load hfiles replication.

Throws:

IOException

getAsyncConnection

private AsyncConnection getAsyncConnection()
                                    throws IOException

Return the shared AsyncConnection which is used for handling wal replication.

Throws:

IOException

getStats

public String getStats()

Get a string representation of this sink's metrics

Returns:

string with the total replicated edits count and the date of the last edit that was applied

getSinkMetrics

public MetricsSink getSinkMetrics()

Get replication Sink Metrics