WALProcedureStore (Apache HBase 1.2.12 API)

java.lang.Object
- org.apache.hadoop.hbase.procedure2.store.ProcedureStoreBase
- - org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore

All Implemented Interfaces:: ProcedureStore

@InterfaceAudience.Private
@InterfaceStability.Evolving
public class WALProcedureStore
extends ProcedureStoreBase

WAL implementation of the ProcedureStore.

Nested Class Summary

Nested Classes
Modifier and Type Class and Description

static interface WALProcedureStore.LeaseRecovery

private static class WALProcedureStore.PushType
- Nested classes/interfaces inherited from interface org.apache.hadoop.hbase.procedure2.store.ProcedureStore
  ProcedureStore.ProcedureIterator, ProcedureStore.ProcedureLoader, ProcedureStore.ProcedureStoreListener

Nested Classes
Modifier and Type	Class and Description
`static interface`	`WALProcedureStore.LeaseRecovery`
`private static class`	`WALProcedureStore.PushType`

Field Summary

Fields
Modifier and Type	Field and Description
`private org.apache.hadoop.conf.Configuration`	`conf`
`private Set<ProcedureWALFile>`	`corruptedLogs`
`private static int`	`DEFAULT_MAX_RETRIES_BEFORE_ROLL`
`private static int`	`DEFAULT_MAX_SYNC_FAILURE_ROLL`
`private static int`	`DEFAULT_PERIODIC_ROLL`
`private static int`	`DEFAULT_ROLL_RETRIES`
`private static long`	`DEFAULT_ROLL_THRESHOLD`
`private static int`	`DEFAULT_SYNC_WAIT_MSEC`
`private static boolean`	`DEFAULT_USE_HSYNC`
`private static int`	`DEFAULT_WAIT_BEFORE_ROLL`
`private static Comparator<org.apache.hadoop.fs.FileStatus>`	`FILE_STATUS_ID_COMPARATOR`
`private long`	`flushLogId`
`private org.apache.hadoop.fs.FileSystem`	`fs`
`private AtomicBoolean`	`inSync`
`private AtomicLong`	`lastRollTs`
`private WALProcedureStore.LeaseRecovery`	`leaseRecovery`
`private AtomicBoolean`	`loading`
`private ReentrantLock`	`lock`
`private static org.apache.commons.logging.Log`	`LOG`
`private org.apache.hadoop.fs.Path`	`logDir`
`private LinkedList<ProcedureWALFile>`	`logs`
`static String`	`MAX_RETRIES_BEFORE_ROLL_CONF_KEY`
`static String`	`MAX_SYNC_FAILURE_ROLL_CONF_KEY`
`private int`	`maxRetriesBeforeRoll`
`private int`	`maxSyncFailureRoll`
`static String`	`PERIODIC_ROLL_CONF_KEY`
`private int`	`periodicRollMsec`
`static String`	`ROLL_RETRIES_CONF_KEY`
`static String`	`ROLL_THRESHOLD_CONF_KEY`
`private int`	`rollRetries`
`private long`	`rollThreshold`
`private Condition`	`slotCond`
`private int`	`slotIndex`
`private ByteSlot[]`	`slots`
`private LinkedTransferQueue<ByteSlot>`	`slotsCache`
`private ProcedureStoreTracker`	`storeTracker`
`private org.apache.hadoop.fs.FSDataOutputStream`	`stream`
`static String`	`SYNC_WAIT_MSEC_CONF_KEY`
`private Condition`	`syncCond`
`private AtomicReference<Throwable>`	`syncException`
`private Thread`	`syncThread`
`private int`	`syncWaitMsec`
`private AtomicLong`	`totalSynced`
`static String`	`USE_HSYNC_CONF_KEY`
`private boolean`	`useHsync`
`static String`	`WAIT_BEFORE_ROLL_CONF_KEY`
`private int`	`waitBeforeRoll`
`private Condition`	`waitCond`
`private static org.apache.hadoop.fs.PathFilter`	`WALS_PATH_FILTER`

Constructor Summary

Constructors
Constructor and Description
`WALProcedureStore(org.apache.hadoop.conf.Configuration conf, org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path logDir, WALProcedureStore.LeaseRecovery leaseRecovery)`

Method Summary

Methods
Modifier and Type	Method and Description
`private ByteSlot`	`acquireSlot()`
`private void`	`closeStream()`
`void`	`delete(long procId)` The specified procId was removed from the executor, due to completion, abort or failure.
`ArrayList<ProcedureWALFile>`	`getActiveLogs()`
`Set<ProcedureWALFile>`	`getCorruptedLogs()`
`org.apache.hadoop.fs.FileSystem`	`getFileSystem()`
`org.apache.hadoop.fs.Path`	`getLogDir()`
`protected org.apache.hadoop.fs.Path`	`getLogFilePath(long logId)`
`private org.apache.hadoop.fs.FileStatus[]`	`getLogFiles()`
`private static long`	`getLogIdFromName(String name)`
`private static long`	`getMaxLogId(org.apache.hadoop.fs.FileStatus[] logFiles)`
`private long`	`getMillisFromLastRoll()`
`private long`	`getMillisToNextPeriodicRoll()`
`int`	`getNumThreads()`
`ProcedureStoreTracker`	`getStoreTracker()`
`private ProcedureWALFile`	`initOldLog(org.apache.hadoop.fs.FileStatus logFile)`
`private long`	`initOldLogs(org.apache.hadoop.fs.FileStatus[] logFiles)`
`private void`	`initTrackerFromOldLogs()`
`void`	`insert(Procedure proc, Procedure[] subprocs)` When a procedure is submitted to the executor insert(proc, null) will be called.
`private boolean`	`isSyncAborted()`
`void`	`load(ProcedureStore.ProcedureLoader loader)` Load the Procedures in the store.
`private void`	`periodicRoll()`
`protected void`	`periodicRollForTesting()`
`private long`	`pushData(WALProcedureStore.PushType type, ByteSlot slot, long procId, long[] subProcIds)`
`void`	`recoverLease()` Acquire the lease for the procedure store.
`private void`	`releaseSlot(ByteSlot slot)`
`private void`	`removeAllLogs(long lastLogId)`
`private void`	`removeInactiveLogs()`
`private boolean`	`removeLogFile(ProcedureWALFile log)`
`private boolean`	`rollWriter()`
`private boolean`	`rollWriter(long logId)`
`protected boolean`	`rollWriterForTesting()`
`private boolean`	`rollWriterOrDie()`
`private void`	`sendStopSignal()`
`void`	`start(int numSlots)` Start/Open the procedure store
`void`	`stop(boolean abort)` Stop/Close the procedure store
`private void`	`syncLoop()`
`private long`	`syncSlots()`
`protected long`	`syncSlots(org.apache.hadoop.fs.FSDataOutputStream stream, ByteSlot[] slots, int offset, int count)`
`private boolean`	`tryRollWriter()`
`void`	`update(Procedure proc)` The specified procedure was executed, and the new state should be written to the store.
`private void`	`updateStoreTracker(WALProcedureStore.PushType type, long procId, long[] subProcIds)`

Methods inherited from class org.apache.hadoop.hbase.procedure2.store.ProcedureStoreBase
isRunning, registerListener, sendAbortProcessSignal, sendPostSyncSignal, setRunning, unregisterListener

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

LOG

private static final org.apache.commons.logging.Log LOG

MAX_RETRIES_BEFORE_ROLL_CONF_KEY

public static final String MAX_RETRIES_BEFORE_ROLL_CONF_KEY

See Also:: Constant Field Values

DEFAULT_MAX_RETRIES_BEFORE_ROLL
```
private static final int DEFAULT_MAX_RETRIES_BEFORE_ROLL
```
See Also:
Constant Field Values

WAIT_BEFORE_ROLL_CONF_KEY

public static final String WAIT_BEFORE_ROLL_CONF_KEY

See Also:: Constant Field Values

DEFAULT_WAIT_BEFORE_ROLL
```
private static final int DEFAULT_WAIT_BEFORE_ROLL
```
See Also:
Constant Field Values

ROLL_RETRIES_CONF_KEY

public static final String ROLL_RETRIES_CONF_KEY

See Also:: Constant Field Values

DEFAULT_ROLL_RETRIES
```
private static final int DEFAULT_ROLL_RETRIES
```
See Also:
Constant Field Values

MAX_SYNC_FAILURE_ROLL_CONF_KEY

public static final String MAX_SYNC_FAILURE_ROLL_CONF_KEY

See Also:: Constant Field Values

DEFAULT_MAX_SYNC_FAILURE_ROLL
```
private static final int DEFAULT_MAX_SYNC_FAILURE_ROLL
```
See Also:
Constant Field Values

PERIODIC_ROLL_CONF_KEY

public static final String PERIODIC_ROLL_CONF_KEY

See Also:: Constant Field Values

DEFAULT_PERIODIC_ROLL
```
private static final int DEFAULT_PERIODIC_ROLL
```
See Also:
Constant Field Values

SYNC_WAIT_MSEC_CONF_KEY

public static final String SYNC_WAIT_MSEC_CONF_KEY

See Also:: Constant Field Values

DEFAULT_SYNC_WAIT_MSEC
```
private static final int DEFAULT_SYNC_WAIT_MSEC
```
See Also:
Constant Field Values

USE_HSYNC_CONF_KEY

public static final String USE_HSYNC_CONF_KEY

See Also:: Constant Field Values

DEFAULT_USE_HSYNC

private static final boolean DEFAULT_USE_HSYNC

See Also:: Constant Field Values

ROLL_THRESHOLD_CONF_KEY

public static final String ROLL_THRESHOLD_CONF_KEY

See Also:: Constant Field Values

DEFAULT_ROLL_THRESHOLD
```
private static final long DEFAULT_ROLL_THRESHOLD
```
See Also:
Constant Field Values

logs

private final LinkedList<ProcedureWALFile> logs

storeTracker

private final ProcedureStoreTracker storeTracker

lock
```
private final ReentrantLock lock
```

waitCond
```
private final Condition waitCond
```

slotCond
```
private final Condition slotCond
```

syncCond
```
private final Condition syncCond
```

leaseRecovery

private final WALProcedureStore.LeaseRecovery leaseRecovery

conf

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

fs

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

logDir

private final org.apache.hadoop.fs.Path logDir

syncException

private final AtomicReference<Throwable> syncException

loading
```
private final AtomicBoolean loading
```

inSync
```
private final AtomicBoolean inSync
```

totalSynced
```
private final AtomicLong totalSynced
```

lastRollTs
```
private final AtomicLong lastRollTs
```

slotsCache

private LinkedTransferQueue<ByteSlot> slotsCache

corruptedLogs

private Set<ProcedureWALFile> corruptedLogs

stream

private org.apache.hadoop.fs.FSDataOutputStream stream

flushLogId
```
private long flushLogId
```

slotIndex
```
private int slotIndex
```

syncThread
```
private Thread syncThread
```

slots
```
private ByteSlot[] slots
```

maxRetriesBeforeRoll
```
private int maxRetriesBeforeRoll
```

maxSyncFailureRoll
```
private int maxSyncFailureRoll
```

waitBeforeRoll
```
private int waitBeforeRoll
```

rollRetries
```
private int rollRetries
```

periodicRollMsec
```
private int periodicRollMsec
```

rollThreshold
```
private long rollThreshold
```

useHsync
```
private boolean useHsync
```

syncWaitMsec
```
private int syncWaitMsec
```

WALS_PATH_FILTER

private static final org.apache.hadoop.fs.PathFilter WALS_PATH_FILTER

FILE_STATUS_ID_COMPARATOR

private static final Comparator<org.apache.hadoop.fs.FileStatus> FILE_STATUS_ID_COMPARATOR

Constructor Detail

WALProcedureStore

public WALProcedureStore(org.apache.hadoop.conf.Configuration conf,
                 org.apache.hadoop.fs.FileSystem fs,
                 org.apache.hadoop.fs.Path logDir,
                 WALProcedureStore.LeaseRecovery leaseRecovery)

Method Detail

start
```
public void start(int numSlots)
           throws IOException
```
Description copied from interface: ProcedureStore

Start/Open the procedure store

Throws:

IOException

stop
```
public void stop(boolean abort)
```
Description copied from interface: ProcedureStore

Stop/Close the procedure store

Parameters:
abort - true if the stop is an abort

sendStopSignal
```
private void sendStopSignal()
```

getNumThreads
```
public int getNumThreads()
```
Returns:
the number of threads/slots passed to start()

getStoreTracker

public ProcedureStoreTracker getStoreTracker()

getActiveLogs

public ArrayList<ProcedureWALFile> getActiveLogs()

getCorruptedLogs

public Set<ProcedureWALFile> getCorruptedLogs()

recoverLease
```
public void recoverLease()
                  throws IOException
```
Description copied from interface: ProcedureStore

Acquire the lease for the procedure store.

Throws:

IOException

load
```
public void load(ProcedureStore.ProcedureLoader loader)
          throws IOException
```
Description copied from interface: ProcedureStore

Load the Procedures in the store.

Parameters:
loader - the ProcedureLoader that will handle the store-load events

Throws:

IOException

insert
```
public void insert(Procedure proc,
          Procedure[] subprocs)
```
Description copied from interface: ProcedureStore

When a procedure is submitted to the executor insert(proc, null) will be called. 'proc' has a 'RUNNABLE' state and the initial information required to start up. When a procedure is executed and it returns children insert(proc, subprocs) will be called. 'proc' has a 'WAITING' state and an update state. 'subprocs' are the children in 'RUNNABLE' state with the initial information.

Parameters:
proc - the procedure to serialize and write to the store.
subprocs - the newly created child of the proc.

update
```
public void update(Procedure proc)
```
Description copied from interface: ProcedureStore

The specified procedure was executed, and the new state should be written to the store.

Parameters:
proc - the procedure to serialize and write to the store.

delete
```
public void delete(long procId)
```
Description copied from interface: ProcedureStore

The specified procId was removed from the executor, due to completion, abort or failure. The store implementor should remove all the information about the specified procId.

Parameters:
procId - the ID of the procedure to remove.

acquireSlot
```
private ByteSlot acquireSlot()
```

releaseSlot

private void releaseSlot(ByteSlot slot)

pushData

private long pushData(WALProcedureStore.PushType type,
            ByteSlot slot,
            long procId,
            long[] subProcIds)

updateStoreTracker

private void updateStoreTracker(WALProcedureStore.PushType type,
                      long procId,
                      long[] subProcIds)

isSyncAborted
```
private boolean isSyncAborted()
```

syncLoop

private void syncLoop()
               throws Throwable

Throws:: Throwable

syncSlots

private long syncSlots()
                throws Throwable

Throws:: Throwable

syncSlots

protected long syncSlots(org.apache.hadoop.fs.FSDataOutputStream stream,
             ByteSlot[] slots,
             int offset,
             int count)
                  throws IOException

Throws:: IOException

rollWriterOrDie
```
private boolean rollWriterOrDie()
```

tryRollWriter
```
private boolean tryRollWriter()
```

getMillisToNextPeriodicRoll

private long getMillisToNextPeriodicRoll()

getMillisFromLastRoll
```
private long getMillisFromLastRoll()
```

periodicRollForTesting

protected void periodicRollForTesting()
                               throws IOException

Throws:: IOException

rollWriterForTesting

protected boolean rollWriterForTesting()
                                throws IOException

Throws:: IOException

periodicRoll

private void periodicRoll()
                   throws IOException

Throws:: IOException

rollWriter

private boolean rollWriter()
                    throws IOException

Throws:: IOException

rollWriter

private boolean rollWriter(long logId)
                    throws IOException

Throws:: IOException

closeStream
```
private void closeStream()
```

removeInactiveLogs
```
private void removeInactiveLogs()
```

removeAllLogs

private void removeAllLogs(long lastLogId)

removeLogFile

private boolean removeLogFile(ProcedureWALFile log)

getLogDir

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

getFileSystem

public org.apache.hadoop.fs.FileSystem getFileSystem()

getLogFilePath

protected org.apache.hadoop.fs.Path getLogFilePath(long logId)
                                            throws IOException

Throws:: IOException

getLogIdFromName

private static long getLogIdFromName(String name)

getLogFiles

private org.apache.hadoop.fs.FileStatus[] getLogFiles()
                                               throws IOException

Throws:: IOException

getMaxLogId

private static long getMaxLogId(org.apache.hadoop.fs.FileStatus[] logFiles)

initOldLogs

private long initOldLogs(org.apache.hadoop.fs.FileStatus[] logFiles)
                  throws IOException

Returns:: Max-LogID of the specified log file set
Throws:: IOException

initTrackerFromOldLogs
```
private void initTrackerFromOldLogs()
```

initOldLog

private ProcedureWALFile initOldLog(org.apache.hadoop.fs.FileStatus logFile)
                             throws IOException

Throws:: IOException

Class WALProcedureStore

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.hadoop.hbase.procedure2.store.ProcedureStore

Field Summary

Constructor Summary

Method Summary

Methods inherited from class org.apache.hadoop.hbase.procedure2.store.ProcedureStoreBase

Methods inherited from class java.lang.Object

Field Detail

LOG

MAX_RETRIES_BEFORE_ROLL_CONF_KEY

DEFAULT_MAX_RETRIES_BEFORE_ROLL

WAIT_BEFORE_ROLL_CONF_KEY

DEFAULT_WAIT_BEFORE_ROLL

ROLL_RETRIES_CONF_KEY

DEFAULT_ROLL_RETRIES

MAX_SYNC_FAILURE_ROLL_CONF_KEY

DEFAULT_MAX_SYNC_FAILURE_ROLL

PERIODIC_ROLL_CONF_KEY

DEFAULT_PERIODIC_ROLL

SYNC_WAIT_MSEC_CONF_KEY

DEFAULT_SYNC_WAIT_MSEC

USE_HSYNC_CONF_KEY

DEFAULT_USE_HSYNC

ROLL_THRESHOLD_CONF_KEY

DEFAULT_ROLL_THRESHOLD

logs

storeTracker

lock

waitCond

slotCond

syncCond

leaseRecovery

conf

fs

logDir

syncException

loading

inSync

totalSynced

lastRollTs

slotsCache

corruptedLogs

stream

flushLogId

slotIndex

syncThread

slots

maxRetriesBeforeRoll

maxSyncFailureRoll

waitBeforeRoll

rollRetries

periodicRollMsec

rollThreshold

useHsync

syncWaitMsec

WALS_PATH_FILTER

FILE_STATUS_ID_COMPARATOR

Constructor Detail

WALProcedureStore

Method Detail

start

stop

sendStopSignal

getNumThreads

getStoreTracker

getActiveLogs

getCorruptedLogs

recoverLease

load

insert

update

delete

acquireSlot

releaseSlot

pushData

updateStoreTracker

isSyncAborted

syncLoop

syncSlots