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

Class ReplicationSourceWALReader

java.lang.Object

java.lang.Thread

org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceWALReader

All Implemented Interfaces:

Runnable

Direct Known Subclasses:

SerialReplicationSourceWALReader

@Private
@Evolving
class ReplicationSourceWALReader
extends Thread

Reads and filters WAL entries, groups the filtered entries into batches, and puts the batches onto a queue

Nested Class Summary

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

Thread.State, Thread.UncaughtExceptionHandler

Field Summary

Fields

Modifier and Type	Field	Description
private final org.apache.hadoop.conf.Configuration	conf
private long	currentPosition
(package private) final BlockingQueue<WALEntryBatch>	entryBatchQueue
private final WALEntryFilter	filter
private final org.apache.hadoop.fs.FileSystem	fs
private boolean	isReaderRunning
private static final org.slf4j.Logger	LOG
private final ReplicationSourceLogQueue	logQueue
private final int	maxRetriesMultiplier
private final int	replicationBatchCountCapacity
private final long	replicationBatchSizeCapacity
private final long	sleepForRetries
private final ReplicationSource	source
(package private) AtomicBoolean	waitingPeerEnabled
private final String	walGroupId

Fields inherited from class java.lang.Thread

MAX_PRIORITY, MIN_PRIORITY, NORM_PRIORITY

Constructor Summary

Constructors

Constructor	Description
ReplicationSourceWALReader(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.conf.Configuration conf, ReplicationSourceLogQueue logQueue, long startPosition, WALEntryFilter filter, ReplicationSource source, String walGroupId)	Creates a reader worker for a given WAL queue.

Method Summary

Modifier and Type	Method	Description
protected final boolean	addEntryToBatch(WALEntryBatch batch, WAL.Entry entry)
private boolean	checkBufferQuota()
private Pair<Integer,Integer>	countDistinctRowKeysAndHFiles(WALEdit edit)	Count the number of different row keys in the given edit because of mini-batching.
private WALEntryBatch	createBatch(WALEntryStream entryStream)
protected final WAL.Entry	filterEntry(WAL.Entry entry)
org.apache.hadoop.fs.Path	getCurrentPath()
private long	getEntrySizeIncludeBulkLoad(WAL.Entry entry)
private ReplicationSourceManager	getSourceManager()
boolean	isReaderRunning()	Returns whether the reader thread is running
WALEntryBatch	poll(long timeout)
protected void	readWALEntries(WALEntryStream entryStream, WALEntryBatch batch)
private void	replicationDone()
void	run()
void	setReaderRunning(boolean readerRunning)
private int	sizeOfStoreFilesIncludeBulkLoad(WALEdit edit)	Calculate the total size of all the store files
protected final int	sleep(int sleepMultiplier)
protected static final boolean	switched(WALEntryStream entryStream, org.apache.hadoop.fs.Path path)
WALEntryBatch	take()	Retrieves the next batch of WAL entries from the queue, waiting up to the specified time for a batch to become available
private void	updateBatchStats(WALEntryBatch batch, WAL.Entry entry, long entrySize)
private void	updateReplicationMarkerEdit(WAL.Entry entry, long offset)

Methods inherited from class java.lang.Thread

activeCount, checkAccess, clone, countStackFrames, currentThread, destroy, dumpStack, enumerate, getAllStackTraces, getContextClassLoader, getDefaultUncaughtExceptionHandler, getId, getName, getPriority, getStackTrace, getState, getThreadGroup, getUncaughtExceptionHandler, 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 Details

LOG

private static final org.slf4j.Logger LOG

logQueue

private final ReplicationSourceLogQueue logQueue

fs

private final org.apache.hadoop.fs.FileSystem fs

conf

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

filter

private final WALEntryFilter filter

source

private final ReplicationSource source

entryBatchQueue

@Private
final BlockingQueue<WALEntryBatch> entryBatchQueue

replicationBatchSizeCapacity

private final long replicationBatchSizeCapacity

replicationBatchCountCapacity

private final int replicationBatchCountCapacity

currentPosition

private long currentPosition

sleepForRetries

private final long sleepForRetries

maxRetriesMultiplier

private final int maxRetriesMultiplier

isReaderRunning

private boolean isReaderRunning

walGroupId

private final String walGroupId

waitingPeerEnabled

AtomicBoolean waitingPeerEnabled

Constructor Details

ReplicationSourceWALReader

public ReplicationSourceWALReader(org.apache.hadoop.fs.FileSystem fs,
 org.apache.hadoop.conf.Configuration conf,
 ReplicationSourceLogQueue logQueue,
 long startPosition,
 WALEntryFilter filter,
 ReplicationSource source,
 String walGroupId)

Creates a reader worker for a given WAL queue. Reads WAL entries off a given queue, batches the entries, and puts them on a batch queue.

Parameters:

fs - the files system to use

conf - configuration to use

logQueue - The WAL queue to read off of

startPosition - position in the first WAL to start reading from

filter - The filter to use while reading

source - replication source

Method Details

replicationDone

private void replicationDone()
                      throws InterruptedException

Throws:

InterruptedException

sleep

protected final int sleep(int sleepMultiplier)

run

public void run()

Specified by:

run in interface Runnable

Overrides:

run in class Thread

addEntryToBatch

protected final boolean addEntryToBatch(WALEntryBatch batch,
 WAL.Entry entry)

switched

protected static final boolean switched(WALEntryStream entryStream,
 org.apache.hadoop.fs.Path path)

readWALEntries

protected void readWALEntries(WALEntryStream entryStream,
 WALEntryBatch batch)
                       throws InterruptedException

Throws:

InterruptedException

getCurrentPath

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

checkBufferQuota

private boolean checkBufferQuota()

createBatch

private WALEntryBatch createBatch(WALEntryStream entryStream)

filterEntry

protected final WAL.Entry filterEntry(WAL.Entry entry)

take

public WALEntryBatch take()
                   throws InterruptedException

Retrieves the next batch of WAL entries from the queue, waiting up to the specified time for a batch to become available

Returns:

A batch of entries, along with the position in the log after reading the batch

Throws:

InterruptedException - if interrupted while waiting

poll

public WALEntryBatch poll(long timeout)
                   throws InterruptedException

Throws:

InterruptedException

getEntrySizeIncludeBulkLoad

private long getEntrySizeIncludeBulkLoad(WAL.Entry entry)

updateBatchStats

private void updateBatchStats(WALEntryBatch batch,
 WAL.Entry entry,
 long entrySize)

countDistinctRowKeysAndHFiles

private Pair<Integer,Integer> countDistinctRowKeysAndHFiles(WALEdit edit)

Count the number of different row keys in the given edit because of mini-batching. We assume that there's at least one Cell in the WALEdit.

Parameters:

edit - edit to count row keys from

Returns:

number of different row keys and HFiles

sizeOfStoreFilesIncludeBulkLoad

private int sizeOfStoreFilesIncludeBulkLoad(WALEdit edit)

Calculate the total size of all the store files

Parameters:

edit - edit to count row keys from

Returns:

the total size of the store files

updateReplicationMarkerEdit

private void updateReplicationMarkerEdit(WAL.Entry entry,
 long offset)

isReaderRunning

public boolean isReaderRunning()

Returns whether the reader thread is running

setReaderRunning

public void setReaderRunning(boolean readerRunning)

Parameters:

readerRunning - the readerRunning to set

getSourceManager

private ReplicationSourceManager getSourceManager()