AsyncProcess (Apache HBase 1.2.12 API)

java.lang.Object
- org.apache.hadoop.hbase.client.AsyncProcess

```
@InterfaceAudience.Private
 class AsyncProcess
extends Object
```
This class allows a continuous flow of requests. It's written to be compatible with a synchronous caller such as HTable.
The caller sends a buffer of operation, by calling submit. This class extract from this list the operations it can send, i.e. the operations that are on region that are not considered as busy. The process is asynchronous, i.e. it returns immediately when if has finished to iterate on the list. If, and only if, the maximum number of current task is reached, the call to submit will block. Alternatively, the caller can call submitAll, in which case all the operations will be sent. Each call to submit returns a future-like object that can be used to track operation progress.

The class manages internally the retries.

The class can be constructed in regular mode, or "global error" mode. In global error mode, AP tracks errors across all calls (each "future" also has global view of all errors). That mode is necessary for backward compat with HTable behavior, where multiple submissions are made and the errors can propagate using any put/flush call, from previous calls. In "regular" mode, the errors are tracked inside the Future object that is returned. The results are always tracked inside the Future object and can be retrieved when the call has finished. Partial results can also be retrieved if some part of multi-request failed.

This class is thread safe in regular mode; in global error code, submitting operations and retrieving errors from different threads may be not thread safe. Internally, the class is thread safe enough to manage simultaneously new submission and results arising from older operations.

Internally, this class works with Row, this mean it could be theoretically used for gets as well.

Nested Class Summary

Nested Classes
Modifier and Type	Class and Description
`static interface`	`AsyncProcess.AsyncRequestFuture` The context used to wait for results from one submit call.
`protected class`	`AsyncProcess.AsyncRequestFutureImpl<CResult>` The context, and return value, for a single submit/submitAll call.
`protected static class`	`AsyncProcess.BatchErrors`
`private static class`	`AsyncProcess.ReplicaResultState` Sync point for calls to multiple replicas for the same user request (Get).
`private static class`	`AsyncProcess.Retry` For manageError.

Field Summary

Fields
Modifier and Type	Field and Description
`protected ClusterConnection`	`connection`
`protected static AtomicLong`	`COUNTER`
`static int`	`DEFAULT_START_LOG_ERRORS_AFTER_COUNT`
`private static int`	`DEFAULT_THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS`
`protected AsyncProcess.BatchErrors`	`globalErrors`
`protected long`	`id`
`private static org.apache.commons.logging.Log`	`LOG`
`static String`	`LOG_DETAILS_FOR_BATCH_ERROR` Configuration to decide whether to log details for batch error
`protected int`	`maxConcurrentTasksPerRegion` The number of tasks we run in parallel on a single region.
`protected int`	`maxConcurrentTasksPerServer` The number of task simultaneously executed on a single region server.
`protected int`	`maxTotalConcurrentTasks` The number of tasks simultaneously executed on the cluster.
`private static AsyncProcess.AsyncRequestFuture`	`NO_REQS_RESULT` Return value from a submit that didn't contain any requests.
`protected int`	`numTries`
`protected long`	`pause`
`protected ExecutorService`	`pool`
`static String`	`PRIMARY_CALL_TIMEOUT_KEY`
`protected long`	`primaryCallTimeoutMicroseconds`
`protected RpcRetryingCallerFactory`	`rpcCallerFactory`
`protected RpcControllerFactory`	`rpcFactory`
`protected int`	`serverTrackerTimeout`
`static String`	`START_LOG_ERRORS_AFTER_COUNT_KEY` Configure the number of failures after which the client will start logging.
`private int`	`startLogErrorsCnt`
`protected ConcurrentMap<byte[],AtomicInteger>`	`taskCounterPerRegion`
`protected ConcurrentMap<ServerName,AtomicInteger>`	`taskCounterPerServer`
`protected AtomicLong`	`tasksInProgress`
`private int`	`THRESHOLD_TO_LOG_REGION_DETAILS`
`private static String`	`THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS`
`private int`	`thresholdToLogUndoneTaskDetails`
`protected int`	`timeout`

Constructor Summary

Constructors
Constructor and Description
`AsyncProcess(ClusterConnection hc, org.apache.hadoop.conf.Configuration conf, ExecutorService pool, RpcRetryingCallerFactory rpcCaller, boolean useGlobalErrors, RpcControllerFactory rpcFactory)`

Method Summary

Methods
Modifier and Type	Method and Description
`private static void`	`addAction(ServerName server, byte[] regionName, Action<Row> action, Map<ServerName,MultiAction<Row>> actionsByServer, long nonceGroup)` Helper that is used when grouping the actions per region server.
`protected boolean`	`canTakeOperation(HRegionLocation loc, Map<HRegionInfo,Boolean> regionsIncluded, Map<ServerName,Boolean> serversIncluded)` Check if we should send new operations to this region or region server.
`protected <CResult> AsyncProcess.AsyncRequestFutureImpl<CResult>`	`createAsyncRequestFuture(TableName tableName, List<Action<Row>> actions, long nonceGroup, ExecutorService pool, Batch.Callback<CResult> callback, Object[] results, boolean needResults)`
`protected MultiServerCallable<Row>`	`createCallable(ServerName server, TableName tableName, MultiAction<Row> multi)` Create a callable.
`protected RpcRetryingCaller<MultiResponse>`	`createCaller(MultiServerCallable<Row> callable)` Create a caller.
`protected ConnectionManager.ServerErrorTracker`	`createServerErrorTracker()` Creates the server error tracker to use inside process.
`protected void`	`decTaskCounters(Collection<byte[]> regions, ServerName sn)` Decrements the counters for a given region and the region server.
`private ExecutorService`	`getPool(ExecutorService pool)`
`boolean`	`hasError()` Only used w/useGlobalErrors ctor argument, for HTable backward compat.
`protected void`	`incTaskCounters(Collection<byte[]> regions, ServerName sn)` increment the tasks counters for a given set of regions.
`private static boolean`	`isReplicaGet(Row row)`
`private void`	`logDetailsOfUndoneTasks(long taskInProgress)`
`private static void`	`setNonce(NonceGenerator ng, Row r, Action<Row> action)`
`<CResult> AsyncProcess.AsyncRequestFuture`	`submit(ExecutorService pool, TableName tableName, List<? extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults)` Extract from the rows list what we can submit.
`<CResult> AsyncProcess.AsyncRequestFuture`	`submit(TableName tableName, List<? extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults)` See `submit(ExecutorService, TableName, List, boolean, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback, boolean)`.
`<CResult> AsyncProcess.AsyncRequestFuture`	`submitAll(ExecutorService pool, TableName tableName, List<? extends Row> rows, Batch.Callback<CResult> callback, Object[] results)` Submit immediately the list of rows, whatever the server status.
`<CResult> AsyncProcess.AsyncRequestFuture`	`submitAll(TableName tableName, List<? extends Row> rows, Batch.Callback<CResult> callback, Object[] results)` See `submitAll(ExecutorService, TableName, List, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback, Object[])`.
`(package private) <CResult> AsyncProcess.AsyncRequestFuture`	`submitMultiActions(TableName tableName, List<Action<Row>> retainedActions, long nonceGroup, Batch.Callback<CResult> callback, Object[] results, boolean needResults, List<Exception> locationErrors, List<Integer> locationErrorRows, Map<ServerName,MultiAction<Row>> actionsByServer, ExecutorService pool)`
`RetriesExhaustedWithDetailsException`	`waitForAllPreviousOpsAndReset(List<Row> failedRows, String tableName)` Only used w/useGlobalErrors ctor argument, for HTable backward compat.
`(package private) void`	`waitForMaximumCurrentTasks(int max, AtomicLong tasksInProgress, long id, String tableName)`
`private void`	`waitForMaximumCurrentTasks(int max, String tableName)` Wait until the async does not have more than max tasks in progress.
`(package private) void`	`waitUntilDone()`

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
```
- COUNTER
```
protected static final AtomicLong COUNTER
```
- PRIMARY_CALL_TIMEOUT_KEY
```
public static final String PRIMARY_CALL_TIMEOUT_KEY
```
  See Also:
  Constant Field Values
- START_LOG_ERRORS_AFTER_COUNT_KEY
```
public static final String START_LOG_ERRORS_AFTER_COUNT_KEY
```
  Configure the number of failures after which the client will start logging. A few failures is fine: region moved, then is not opened, then is overloaded. We try to have an acceptable heuristic for the number of errors we don't log. 9 was chosen because we wait for 1s at this stage.
  
  See Also:
  Constant Field Values
- DEFAULT_START_LOG_ERRORS_AFTER_COUNT
```
public static final int DEFAULT_START_LOG_ERRORS_AFTER_COUNT
```
  See Also:
  Constant Field Values
- LOG_DETAILS_FOR_BATCH_ERROR
```
public static final String LOG_DETAILS_FOR_BATCH_ERROR
```
  Configuration to decide whether to log details for batch error
  
  See Also:
  Constant Field Values
- thresholdToLogUndoneTaskDetails
```
private final int thresholdToLogUndoneTaskDetails
```
- THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS
```
private static final String THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS
```
  See Also:
  Constant Field Values
- DEFAULT_THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS
```
private static final int DEFAULT_THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS
```
  See Also:
  Constant Field Values
- THRESHOLD_TO_LOG_REGION_DETAILS
```
private final int THRESHOLD_TO_LOG_REGION_DETAILS
```
  See Also:
  Constant Field Values
- NO_REQS_RESULT
```
private static final AsyncProcess.AsyncRequestFuture NO_REQS_RESULT
```
  Return value from a submit that didn't contain any requests.
- id
```
protected final long id
```
- connection
```
protected final ClusterConnection connection
```
- rpcCallerFactory
```
protected final RpcRetryingCallerFactory rpcCallerFactory
```
- rpcFactory
```
protected final RpcControllerFactory rpcFactory
```
- globalErrors
```
protected final AsyncProcess.BatchErrors globalErrors
```
- pool
```
protected final ExecutorService pool
```
- tasksInProgress
```
protected final AtomicLong tasksInProgress
```
- taskCounterPerRegion
```
protected final ConcurrentMap<byte[],AtomicInteger> taskCounterPerRegion
```
- taskCounterPerServer
```
protected final ConcurrentMap<ServerName,AtomicInteger> taskCounterPerServer
```
- startLogErrorsCnt
```
private final int startLogErrorsCnt
```
- maxTotalConcurrentTasks
```
protected final int maxTotalConcurrentTasks
```
  The number of tasks simultaneously executed on the cluster.
- maxConcurrentTasksPerRegion
```
protected final int maxConcurrentTasksPerRegion
```
  The number of tasks we run in parallel on a single region. With 1 (the default) , we ensure that the ordering of the queries is respected: we don't start a set of operations on a region before the previous one is done. As well, this limits the pressure we put on the region server.
- maxConcurrentTasksPerServer
```
protected final int maxConcurrentTasksPerServer
```
  The number of task simultaneously executed on a single region server.
- pause
```
protected final long pause
```
- numTries
```
protected int numTries
```
- serverTrackerTimeout
```
protected int serverTrackerTimeout
```
- timeout
```
protected int timeout
```
- primaryCallTimeoutMicroseconds
```
protected long primaryCallTimeoutMicroseconds
```

Constructor Detail

AsyncProcess

public AsyncProcess(ClusterConnection hc,
            org.apache.hadoop.conf.Configuration conf,
            ExecutorService pool,
            RpcRetryingCallerFactory rpcCaller,
            boolean useGlobalErrors,
            RpcControllerFactory rpcFactory)

Method Detail

getPool
```
private ExecutorService getPool(ExecutorService pool)
```
Returns:
pool if non null, otherwise returns this.pool if non null, otherwise throws RuntimeException

submit

public <CResult> AsyncProcess.AsyncRequestFuture submit(TableName tableName,
                                               List<? extends Row> rows,
                                               boolean atLeastOne,
                                               Batch.Callback<CResult> callback,
                                               boolean needResults)
                                       throws InterruptedIOException

See submit(ExecutorService, TableName, List, boolean, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback, boolean). Uses default ExecutorService for this AP (must have been created with one).

Throws:: InterruptedIOException

submit

public <CResult> AsyncProcess.AsyncRequestFuture submit(ExecutorService pool,
                                               TableName tableName,
                                               List<? extends Row> rows,
                                               boolean atLeastOne,
                                               Batch.Callback<CResult> callback,
                                               boolean needResults)
                                       throws InterruptedIOException

Extract from the rows list what we can submit. The rows we can not submit are kept in the list. Does not send requests to replicas (not currently used for anything other than streaming puts anyway).

Parameters:: pool - ExecutorService to use.; tableName - The table for which this request is needed.; callback - Batch callback. Only called on success (94 behavior).; needResults - Whether results are needed, or can be discarded.; rows - - the submitted row. Modified by the method: we remove the rows we took.; atLeastOne - true if we should submit at least a subset.
Throws:: InterruptedIOException

submitMultiActions

<CResult> AsyncProcess.AsyncRequestFuture submitMultiActions(TableName tableName,
                                                           List<Action<Row>> retainedActions,
                                                           long nonceGroup,
                                                           Batch.Callback<CResult> callback,
                                                           Object[] results,
                                                           boolean needResults,
                                                           List<Exception> locationErrors,
                                                           List<Integer> locationErrorRows,
                                                           Map<ServerName,MultiAction<Row>> actionsByServer,
                                                           ExecutorService pool)

addAction

private static void addAction(ServerName server,
             byte[] regionName,
             Action<Row> action,
             Map<ServerName,MultiAction<Row>> actionsByServer,
             long nonceGroup)

Helper that is used when grouping the actions per region server.

Parameters:: loc - - the destination. Must not be null.; action - - the action to add to the multiaction; actionsByServer - the multiaction per server; nonceGroup - Nonce group.

canTakeOperation
```
protected boolean canTakeOperation(HRegionLocation loc,
                       Map<HRegionInfo,Boolean> regionsIncluded,
                       Map<ServerName,Boolean> serversIncluded)
```
Check if we should send new operations to this region or region server. We're taking into account the past decision; if we have already accepted operation on a given region, we accept all operations for this region.

Parameters:
loc; - the region and the server name we want to use.

Returns:
true if this region is considered as busy.

submitAll

public <CResult> AsyncProcess.AsyncRequestFuture submitAll(TableName tableName,
                                                  List<? extends Row> rows,
                                                  Batch.Callback<CResult> callback,
                                                  Object[] results)

See submitAll(ExecutorService, TableName, List, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback, Object[]). Uses default ExecutorService for this AP (must have been created with one).

submitAll

public <CResult> AsyncProcess.AsyncRequestFuture submitAll(ExecutorService pool,
                                                  TableName tableName,
                                                  List<? extends Row> rows,
                                                  Batch.Callback<CResult> callback,
                                                  Object[] results)

Submit immediately the list of rows, whatever the server status. Kept for backward compatibility: it allows to be used with the batch interface that return an array of objects.

Parameters:: pool - ExecutorService to use.; tableName - name of the table for which the submission is made.; rows - the list of rows.; callback - the callback.; results - Optional array to return the results thru; backward compat.

setNonce

private static void setNonce(NonceGenerator ng,
            Row r,
            Action<Row> action)

createAsyncRequestFuture

protected <CResult> AsyncProcess.AsyncRequestFutureImpl<CResult> createAsyncRequestFuture(TableName tableName,
                                                                              List<Action<Row>> actions,
                                                                              long nonceGroup,
                                                                              ExecutorService pool,
                                                                              Batch.Callback<CResult> callback,
                                                                              Object[] results,
                                                                              boolean needResults)

createCallable

protected MultiServerCallable<Row> createCallable(ServerName server,
                                      TableName tableName,
                                      MultiAction<Row> multi)

Create a callable. Isolated to be easily overridden in the tests.

createCaller

protected RpcRetryingCaller<MultiResponse> createCaller(MultiServerCallable<Row> callable)

Create a caller. Isolated to be easily overridden in the tests.

waitUntilDone

void waitUntilDone()
             throws InterruptedIOException

Throws:: InterruptedIOException

waitForMaximumCurrentTasks

private void waitForMaximumCurrentTasks(int max,
                              String tableName)
                                 throws InterruptedIOException

Wait until the async does not have more than max tasks in progress.

Throws:: InterruptedIOException

waitForMaximumCurrentTasks

void waitForMaximumCurrentTasks(int max,
                              AtomicLong tasksInProgress,
                              long id,
                              String tableName)
                          throws InterruptedIOException

Throws:: InterruptedIOException

logDetailsOfUndoneTasks

private void logDetailsOfUndoneTasks(long taskInProgress)

hasError
```
public boolean hasError()
```
Only used w/useGlobalErrors ctor argument, for HTable backward compat.

Returns:
Whether there were any errors in any request since the last time #waitForAllPreviousOpsAndReset(List) was called, or AP was created.

waitForAllPreviousOpsAndReset
```
public RetriesExhaustedWithDetailsException waitForAllPreviousOpsAndReset(List<Row> failedRows,
                                                                 String tableName)
                                                                   throws InterruptedIOException
```
Only used w/useGlobalErrors ctor argument, for HTable backward compat. Waits for all previous operations to finish, and returns errors and (optionally) failed operations themselves.

Parameters:
failedRows - an optional list into which the rows that failed since the last time #waitForAllPreviousOpsAndReset(List) was called, or AP was created, are saved.
tableName - name of the table

Returns:
all the errors since the last time #waitForAllPreviousOpsAndReset(List) was called, or AP was created.

Throws:

InterruptedIOException

incTaskCounters

protected void incTaskCounters(Collection<byte[]> regions,
                   ServerName sn)

increment the tasks counters for a given set of regions. MT safe.

decTaskCounters

protected void decTaskCounters(Collection<byte[]> regions,
                   ServerName sn)

Decrements the counters for a given region and the region server. MT Safe.

createServerErrorTracker
```
protected ConnectionManager.ServerErrorTracker createServerErrorTracker()
```
Creates the server error tracker to use inside process. Currently, to preserve the main assumption about current retries, and to work well with the retry-limit-based calculation, the calculation is local per Process object. We may benefit from connection-wide tracking of server errors.

Returns:
ServerErrorTracker to use, null if there is no ServerErrorTracker on this connection

isReplicaGet

private static boolean isReplicaGet(Row row)

Class AsyncProcess

Nested Class Summary

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

LOG

COUNTER

PRIMARY_CALL_TIMEOUT_KEY

START_LOG_ERRORS_AFTER_COUNT_KEY

DEFAULT_START_LOG_ERRORS_AFTER_COUNT

LOG_DETAILS_FOR_BATCH_ERROR

thresholdToLogUndoneTaskDetails

THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS

DEFAULT_THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS

THRESHOLD_TO_LOG_REGION_DETAILS

NO_REQS_RESULT

id

connection

rpcCallerFactory

rpcFactory

globalErrors

pool

tasksInProgress

taskCounterPerRegion

taskCounterPerServer

startLogErrorsCnt

maxTotalConcurrentTasks

maxConcurrentTasksPerRegion

maxConcurrentTasksPerServer

pause

numTries

serverTrackerTimeout

timeout

primaryCallTimeoutMicroseconds

Constructor Detail

AsyncProcess

Method Detail

getPool

submit

submit

submitMultiActions

addAction

canTakeOperation

submitAll

submitAll

setNonce

createAsyncRequestFuture

createCallable

createCaller

waitUntilDone

waitForMaximumCurrentTasks

waitForMaximumCurrentTasks

logDetailsOfUndoneTasks

hasError

waitForAllPreviousOpsAndReset

incTaskCounters

decTaskCounters

createServerErrorTracker

isReplicaGet