org.apache.hadoop.hbase.client

Class AsyncProcess

java.lang.Object

org.apache.hadoop.hbase.client.AsyncProcess

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

Field Summary

Fields

Modifier and Type	Field and Description
(package private) ClusterConnection	connection
(package private) ConnectionConfiguration	connectionConfiguration
private static AtomicLong	COUNTER
private static int	DEFAULT_LOG_DETAILS_PERIOD
static int	DEFAULT_START_LOG_ERRORS_AFTER_COUNT
(package private) long	id
private static org.slf4j.Logger	LOG
static String	LOG_DETAILS_FOR_BATCH_ERROR Configuration to decide whether to log details for batch error
static String	LOG_DETAILS_PERIOD
(package private) boolean	logBatchErrorDetails Whether to log details for batch errors
private static AsyncRequestFuture	NO_REQS_RESULT Return value from a submit that didn't contain any requests.
(package private) int	numTries
private int	periodToLog
static String	PRIMARY_CALL_TIMEOUT_KEY
(package private) long	primaryCallTimeoutMicroseconds
(package private) RequestController	requestController The traffic control for requests.
private RpcRetryingCallerFactory	rpcCallerFactory
(package private) RpcControllerFactory	rpcFactory
(package private) long	serverTrackerTimeout
static String	START_LOG_ERRORS_AFTER_COUNT_KEY Configure the number of failures after which the client will start logging.
(package private) int	startLogErrorsCnt

Constructor Summary

Constructors

Constructor and Description
AsyncProcess(ClusterConnection hc, org.apache.hadoop.conf.Configuration conf, RpcRetryingCallerFactory rpcCaller, RpcControllerFactory rpcFactory)
AsyncProcess(ClusterConnection hc, org.apache.hadoop.conf.Configuration conf, RpcRetryingCallerFactory rpcCaller, RpcControllerFactory rpcFactory, int retriesNumber)

Method Summary

Modifier and Type	Method and Description
(package private) static void	addAction(ServerName server, byte[] regionName, Action action, Map<ServerName,MultiAction> actionsByServer, long nonceGroup) Helper that is used when grouping the actions per region server.
private int	checkOperationTimeout(int operationTimeout)
private int	checkRpcTimeout(int rpcTimeout)
private <CResult> AsyncRequestFuture	checkTask(AsyncProcessTask<CResult> task)
private int	checkTimeout(String name, int timeout)
(package private) <CResult> AsyncRequestFutureImpl<CResult>	createAsyncRequestFuture(AsyncProcessTask task, List<Action> actions, long nonceGroup)
protected RpcRetryingCaller<AbstractResponse>	createCaller(CancellableRegionServerCallable callable, int rpcTimeout) Create a caller.
(package private) ConnectionImplementation.ServerErrorTracker	createServerErrorTracker() Creates the server error tracker to use inside process.
(package private) void	decTaskCounters(Collection<byte[]> regions, ServerName sn)
private Consumer<Long>	getLogger(TableName tableName, long max)
private static boolean	hasIncrementOrAppend(Row action)
private static boolean	hasIncrementOrAppend(RowMutations mutations)
(package private) void	incTaskCounters(Collection<byte[]> regions, ServerName sn)
(package private) static boolean	isReplicaGet(Row row)
private void	setNonce(NonceGenerator ng, Row r, Action action)
<CResult> AsyncRequestFuture	submit(AsyncProcessTask<CResult> task) The submitted task may be not accomplished at all if there are too many running tasks or other limits.
private <CResult> AsyncRequestFuture	submit(AsyncProcessTask<CResult> task, boolean atLeastOne) Extract from the rows list what we can submit.
private <CResult> AsyncRequestFuture	submitAll(AsyncProcessTask task) Submit immediately the list of rows, whatever the server status.
(package private) <CResult> AsyncRequestFuture	submitMultiActions(AsyncProcessTask task, List<Action> retainedActions, long nonceGroup, List<Exception> locationErrors, List<Integer> locationErrorRows, Map<ServerName,MultiAction> actionsByServer)
protected void	waitForMaximumCurrentTasks(int max, TableName tableName) Wait until the async does not have more than max tasks in progress.

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.slf4j.Logger LOG

COUNTER

private 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. 5 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

NO_REQS_RESULT

private static final AsyncRequestFuture NO_REQS_RESULT

Return value from a submit that didn't contain any requests.

id

final long id

connection

final ClusterConnection connection

connectionConfiguration

final ConnectionConfiguration connectionConfiguration

rpcCallerFactory

private final RpcRetryingCallerFactory rpcCallerFactory

rpcFactory

final RpcControllerFactory rpcFactory

startLogErrorsCnt

final int startLogErrorsCnt

numTries

final int numTries

serverTrackerTimeout

long serverTrackerTimeout

primaryCallTimeoutMicroseconds

final long primaryCallTimeoutMicroseconds

logBatchErrorDetails

final boolean logBatchErrorDetails

Whether to log details for batch errors

requestController

final RequestController requestController

The traffic control for requests.

LOG_DETAILS_PERIOD

public static final String LOG_DETAILS_PERIOD

See Also:

Constant Field Values

DEFAULT_LOG_DETAILS_PERIOD

private static final int DEFAULT_LOG_DETAILS_PERIOD

See Also:

Constant Field Values

periodToLog

private final int periodToLog

Constructor Detail

AsyncProcess

AsyncProcess(ClusterConnection hc,
             org.apache.hadoop.conf.Configuration conf,
             RpcRetryingCallerFactory rpcCaller,
             RpcControllerFactory rpcFactory)

AsyncProcess

AsyncProcess(ClusterConnection hc,
             org.apache.hadoop.conf.Configuration conf,
             RpcRetryingCallerFactory rpcCaller,
             RpcControllerFactory rpcFactory,
             int retriesNumber)

Method Detail

submit

public <CResult> AsyncRequestFuture submit(AsyncProcessTask<CResult> task)
                                    throws InterruptedIOException

The submitted task may be not accomplished at all if there are too many running tasks or other limits.

Type Parameters:

CResult - The class to cast the result

Parameters:

task - The setting and data n

Throws:

InterruptedIOException

submit

private <CResult> AsyncRequestFuture submit(AsyncProcessTask<CResult> task,
                                            boolean atLeastOne)
                                     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:

task - The setting and data

atLeastOne - true if we should submit at least a subset.

Throws:

InterruptedIOException

submitMultiActions

<CResult> AsyncRequestFuture submitMultiActions(AsyncProcessTask task,
                                                List<Action> retainedActions,
                                                long nonceGroup,
                                                List<Exception> locationErrors,
                                                List<Integer> locationErrorRows,
                                                Map<ServerName,MultiAction> actionsByServer)

addAction

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

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

Parameters:

server - - server

regionName - - regionName

action - - the action to add to the multiaction

actionsByServer - the multiaction per server

nonceGroup - Nonce group.

submitAll

private <CResult> AsyncRequestFuture submitAll(AsyncProcessTask task)

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:

task - The setting and data

checkTask

private <CResult> AsyncRequestFuture checkTask(AsyncProcessTask<CResult> task)

setNonce

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

hasIncrementOrAppend

private static boolean hasIncrementOrAppend(Row action)

hasIncrementOrAppend

private static boolean hasIncrementOrAppend(RowMutations mutations)

checkTimeout

private int checkTimeout(String name,
                         int timeout)

checkOperationTimeout

private int checkOperationTimeout(int operationTimeout)

checkRpcTimeout

private int checkRpcTimeout(int rpcTimeout)

createAsyncRequestFuture

<CResult> AsyncRequestFutureImpl<CResult> createAsyncRequestFuture(AsyncProcessTask task,
                                                                   List<Action> actions,
                                                                   long nonceGroup)

waitForMaximumCurrentTasks

protected void waitForMaximumCurrentTasks(int max,
                                          TableName tableName)
                                   throws InterruptedIOException

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

Throws:

InterruptedIOException

getLogger

private Consumer<Long> getLogger(TableName tableName,
                                 long max)

incTaskCounters

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

decTaskCounters

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

createCaller

protected RpcRetryingCaller<AbstractResponse> createCaller(CancellableRegionServerCallable callable,
                                                           int rpcTimeout)

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

createServerErrorTracker

ConnectionImplementation.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

static boolean isReplicaGet(Row row)