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:

RecoveredReplicationSourceWALReader

@InterfaceAudience.Private
 @InterfaceStability.Evolving
public 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

Modifier and Type	Class and Description
(package private) static class	ReplicationSourceWALReader.WALEntryBatch Holds a batch of WAL entries to replicate, along with some statistics

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

Thread.State, Thread.UncaughtExceptionHandler

Field Summary

Fields

Modifier and Type	Field and Description
private org.apache.hadoop.conf.Configuration	conf
private long	currentPosition
protected BlockingQueue<ReplicationSourceWALReader.WALEntryBatch>	entryBatchQueue
private boolean	eofAutoRecovery
private WALEntryFilter	filter
private org.apache.hadoop.fs.FileSystem	fs
private boolean	isReaderRunning
private static org.slf4j.Logger	LOG
private PriorityBlockingQueue<org.apache.hadoop.fs.Path>	logQueue
private int	maxRetriesMultiplier
protected int	replicationBatchCountCapacity
private long	replicationBatchSizeCapacity
private long	sleepForRetries
private ReplicationSource	source
private long	totalBufferQuota
private AtomicLong	totalBufferUsed

Fields inherited from class java.lang.Thread

MAX_PRIORITY, MIN_PRIORITY, NORM_PRIORITY

Constructor Summary

Constructors

Constructor and Description
ReplicationSourceWALReader(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.conf.Configuration conf, PriorityBlockingQueue<org.apache.hadoop.fs.Path> logQueue, long startPosition, WALEntryFilter filter, ReplicationSource source) Creates a reader worker for a given WAL queue.

Method Summary

Modifier and Type	Method and Description
private boolean	acquireBufferQuota(long size)
private int	calculateTotalSizeOfStoreFiles(WALEdit edit) Calculate the total size of all the store files
private boolean	checkQuota()
private Pair<Integer,Integer>	countDistinctRowKeysAndHFiles(WALEdit edit) Count the number of different row keys in the given edit because of mini-batching.
private WAL.Entry	filterEntry(WAL.Entry entry)
org.apache.hadoop.fs.Path	getCurrentPath()
private long	getEntrySize(WAL.Entry entry)
protected void	handleEmptyWALEntryBatch(ReplicationSourceWALReader.WALEntryBatch batch, org.apache.hadoop.fs.Path currentPath)
private void	handleEofException(IOException e)
boolean	isReaderRunning()
private ReplicationSourceWALReader.WALEntryBatch	readWALEntries(WALEntryStream entryStream)
void	run()
void	setReaderRunning(boolean readerRunning)
ReplicationSourceWALReader.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(ReplicationSourceWALReader.WALEntryBatch batch, WAL.Entry entry, long entryPosition, long entrySize)

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 Detail

LOG

private static final org.slf4j.Logger LOG

logQueue

private final PriorityBlockingQueue<org.apache.hadoop.fs.Path> 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

protected final BlockingQueue<ReplicationSourceWALReader.WALEntryBatch> entryBatchQueue

replicationBatchSizeCapacity

private final long replicationBatchSizeCapacity

replicationBatchCountCapacity

protected final int replicationBatchCountCapacity

currentPosition

private long currentPosition

sleepForRetries

private final long sleepForRetries

maxRetriesMultiplier

private final int maxRetriesMultiplier

eofAutoRecovery

private final boolean eofAutoRecovery

isReaderRunning

private boolean isReaderRunning

totalBufferUsed

private AtomicLong totalBufferUsed

totalBufferQuota

private long totalBufferQuota

Constructor Detail

ReplicationSourceWALReader

public ReplicationSourceWALReader(org.apache.hadoop.fs.FileSystem fs,
                                  org.apache.hadoop.conf.Configuration conf,
                                  PriorityBlockingQueue<org.apache.hadoop.fs.Path> logQueue,
                                  long startPosition,
                                  WALEntryFilter filter,
                                  ReplicationSource source)

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 Detail

run

public void run()

Specified by:

run in interface Runnable

Overrides:

run in class Thread

readWALEntries

private ReplicationSourceWALReader.WALEntryBatch readWALEntries(WALEntryStream entryStream)
                                                         throws IOException

Throws:

IOException

handleEmptyWALEntryBatch

protected void handleEmptyWALEntryBatch(ReplicationSourceWALReader.WALEntryBatch batch,
                                        org.apache.hadoop.fs.Path currentPath)
                                 throws InterruptedException

Throws:

InterruptedException

handleEofException

private void handleEofException(IOException e)

getCurrentPath

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

checkQuota

private boolean checkQuota()

filterEntry

private WAL.Entry filterEntry(WAL.Entry entry)

take

public ReplicationSourceWALReader.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

getEntrySize

private long getEntrySize(WAL.Entry entry)

updateBatchStats

private void updateBatchStats(ReplicationSourceWALReader.WALEntryBatch batch,
                              WAL.Entry entry,
                              long entryPosition,
                              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

calculateTotalSizeOfStoreFiles

private int calculateTotalSizeOfStoreFiles(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

acquireBufferQuota

private boolean acquireBufferQuota(long size)

Parameters:

size - delta size for grown buffer

Returns:

true if we should clear buffer and push all

isReaderRunning

public boolean isReaderRunning()

Returns:

whether the reader thread is running

setReaderRunning

public void setReaderRunning(boolean readerRunning)

Parameters:

readerRunning - the readerRunning to set