Package org.apache.hadoop.hbase.client

Class AsyncRequestFutureImpl<CResult>

java.lang.Object

org.apache.hadoop.hbase.client.AsyncRequestFutureImpl<CResult>

All Implemented Interfaces:

AsyncRequestFuture

@Private
class AsyncRequestFutureImpl<CResult>
extends Object
implements AsyncRequestFuture

The context, and return value, for a single submit/submitAll call. Note on how this class (one AP submit) works. Initially, all requests are split into groups by server; request is sent to each server in parallel; the RPC calls are not async so a thread per server is used. Every time some actions fail, regions/locations might have changed, so we re-group them by server and region again and send these groups in parallel too. The result, in case of retries, is a "tree" of threads, with parent exiting after scheduling children. This is why lots of code doesn't require any synchronization.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
private final class	AsyncRequestFutureImpl.ReplicaCallIssuingRunnable	Runnable (that can be submitted to thread pool) that waits for when it's time to issue replica calls, finds region replicas, groups the requests by replica and issues the calls (on separate threads, via sendMultiAction).
private static class	AsyncRequestFutureImpl.ReplicaResultState	Sync point for calls to multiple replicas for the same user request (Get).
static enum	AsyncRequestFutureImpl.Retry	For manageError(int, Row, Retry, Throwable, ServerName).
(package private) final class	AsyncRequestFutureImpl.SingleServerRequestRunnable	Runnable (that can be submitted to thread pool) that submits MultiAction to a single server.

Field Summary

Fields

Modifier and Type	Field	Description
private final AtomicLong	actionsInProgress
private final AsyncProcess	asyncProcess
private final Batch.Callback<CResult>	callback
private final Set<CancellableRegionServerCallable>	callsInProgress
private final CancellableRegionServerCallable	currentCallable
private final BatchErrors	errors
private final ConnectionImplementation.ServerErrorTracker	errorsByServer
private final boolean	hasAnyReplicaGets
private static final org.slf4j.Logger	LOG
private final long	nonceGroup
private final int	operationTimeout
private final ExecutorService	pool
private final int[]	replicaGetIndices	Indices of replica gets in results.
private final Object	replicaResultLock	The lock controls access to results.
private final Map<String,byte[]>	requestAttributes
private final Object[]	results	Result array.
private final int	rpcTimeout
private final TableName	tableName
private RetryingTimeTracker	tracker

Constructor Summary

Constructors

Constructor	Description
AsyncRequestFutureImpl(AsyncProcessTask task, List<Action> actions, long nonceGroup, AsyncProcess asyncProcess)

Method Summary

Modifier and Type	Method	Description
private String	buildDetailedErrorMsg(String string, int index)
private void	cleanServerCache(ServerName server, Throwable regionException)
private MultiServerCallable	createCallable(ServerName server, TableName tableName, MultiAction multi)	Create a callable.
private String	createLog(int numAttempt, int failureCount, int replaySize, ServerName sn, Throwable error, long backOffTime, boolean willRetry, String startTime, int failed, int stopped)
private ConnectionImplementation.ServerErrorTracker	createServerErrorTracker()	Creates the server error tracker to use inside process.
(package private) AsyncRequestFutureImpl<CResult>.SingleServerRequestRunnable	createSingleServerRequest(MultiAction multiAction, int numAttempt, ServerName server, Set<CancellableRegionServerCallable> callsInProgress)
private void	decActionCounter(int index)
private void	failAll(MultiAction actions, ServerName server, int numAttempt, Throwable throwable)	Fail all the actions from this multiaction after an OperationTimeoutExceededException
private RegionLocations	findAllLocationsOrFail(Action action, boolean useCache)
private Long	getBackoff(ServerName server, byte[] regionName)
protected Set<CancellableRegionServerCallable>	getCallsInProgress()
RetriesExhaustedWithDetailsException	getErrors()
List<? extends Row>	getFailedOperations()
private Collection<? extends Runnable>	getNewMultiActionRunnable(ServerName server, MultiAction multiAction, int numAttempt)
(package private) long	getNumberOfActionsInProgress()
private long	getRemainingTime()
private HRegionLocation	getReplicaLocationOrFail(Action action)
Object[]	getResults()
(package private) void	groupAndSendMultiAction(List<Action> currentActions, int numAttempt)	Group a list of actions per region servers, and send them.
boolean	hasError()
private void	invokeCallBack(byte[] regionName, byte[] row, CResult result)
private boolean	isActionComplete(int index, Row row)	Checks if the action is complete; used on error to prevent needless retries.
private boolean	isOperationTimeoutExceeded()	Some checked calls send a callable with their own tracker.
private void	logNoResubmit(ServerName oldServer, int numAttempt, int failureCount, Throwable throwable, int failed, int stopped)
(package private) AsyncRequestFutureImpl.Retry	manageError(int originalIndex, Row row, AsyncRequestFutureImpl.Retry canRetry, Throwable throwable, ServerName server)	Check that we can retry acts accordingly: logs, set the error status.
private void	manageLocationError(Action action, Exception ex)
private void	receiveGlobalFailure(MultiAction rsActions, ServerName server, int numAttempt, Throwable t, boolean clearServerCache)	Resubmit all the actions from this multiaction after a failure.
private void	receiveMultiAction(MultiAction multiAction, ServerName server, MultiResponse responses, int numAttempt)	Called when we receive the result of a server query.
private void	resubmit(ServerName oldServer, List<Action> toReplay, int numAttempt, int failureCount, Throwable throwable)	Log as much info as possible, and, if there is something to replay, submit it again after a back off sleep.
(package private) void	sendMultiAction(Map<ServerName,MultiAction> actionsByServer, int numAttempt, List<Action> actionsForReplicaThread, boolean reuseThread)	Send a multi action structure to the servers, after a delay depending on the attempt number.
private void	setError(int index, Row row, Throwable throwable, ServerName server)	Sets the error from a particular action.
private void	setResult(Action action, Object result)	Sets the non-error result from a particular action.
private void	startWaitingForReplicaCalls(List<Action> actionsForReplicaThread)	Starts waiting to issue replica calls on a different thread; or issues them immediately.
private AsyncRequestFutureImpl.ReplicaResultState	trySetResultSimple(int index, Row row, boolean isError, Object result, ServerName server, boolean isFromReplica)	Tries to set the result or error for a particular action as if there were no replica calls.
private void	updateCachedLocations(ServerName server, byte[] regionName, byte[] row, Throwable rowException)
private void	updateResult(int index, Object result)
protected void	updateStats(ServerName server, MultiResponse resp)
void	waitUntilDone()	Wait until all tasks are executed, successfully or not.
private boolean	waitUntilDone(long cutoff)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

LOG

private static final org.slf4j.Logger LOG

tracker

private RetryingTimeTracker tracker

callback

private final Batch.Callback<CResult> callback

errors

private final BatchErrors errors

errorsByServer

private final ConnectionImplementation.ServerErrorTracker errorsByServer

pool

private final ExecutorService pool

callsInProgress

private final Set<CancellableRegionServerCallable> callsInProgress

tableName

private final TableName tableName

actionsInProgress

private final AtomicLong actionsInProgress

replicaResultLock

private final Object replicaResultLock

The lock controls access to results. It is only held when populating results where there might be several callers (eventual consistency gets). For other requests, there's one unique call going on per result index.

results

private final Object[] results

Result array. Null if results are not needed. Otherwise, each index corresponds to the action index in initial actions submitted. For most request types, has null-s for requests that are not done, and result/exception for those that are done. For eventual-consistency gets, initially the same applies; at some point, replica calls might be started, and ReplicaResultState is put at the corresponding indices. The returning calls check the type to detect when this is the case. After all calls are done, ReplicaResultState-s are replaced with results for the user.

replicaGetIndices

private final int[] replicaGetIndices

Indices of replica gets in results. If null, all or no actions are replica-gets.

hasAnyReplicaGets

private final boolean hasAnyReplicaGets

nonceGroup

private final long nonceGroup

currentCallable

private final CancellableRegionServerCallable currentCallable

operationTimeout

private final int operationTimeout

rpcTimeout

private final int rpcTimeout

asyncProcess

private final AsyncProcess asyncProcess

requestAttributes

private final Map<String,byte[]> requestAttributes

Constructor Details

AsyncRequestFutureImpl

public AsyncRequestFutureImpl(AsyncProcessTask task,
 List<Action> actions,
 long nonceGroup,
 AsyncProcess asyncProcess)

Method Details

getCallsInProgress

protected Set<CancellableRegionServerCallable> getCallsInProgress()

createSingleServerRequest

AsyncRequestFutureImpl<CResult>.SingleServerRequestRunnable createSingleServerRequest(MultiAction multiAction,
 int numAttempt,
 ServerName server,
 Set<CancellableRegionServerCallable> callsInProgress)

isOperationTimeoutExceeded

private boolean isOperationTimeoutExceeded()

Some checked calls send a callable with their own tracker. This method checks the operation timeout against the appropriate tracker, or returns false if no tracker.

getRemainingTime

private long getRemainingTime()

groupAndSendMultiAction

void groupAndSendMultiAction(List<Action> currentActions,
 int numAttempt)

Group a list of actions per region servers, and send them.

Parameters:

currentActions - - the list of row to submit

numAttempt - - the current numAttempt (first attempt is 1)

getReplicaLocationOrFail

private HRegionLocation getReplicaLocationOrFail(Action action)

manageLocationError

private void manageLocationError(Action action,
 Exception ex)

findAllLocationsOrFail

private RegionLocations findAllLocationsOrFail(Action action,
 boolean useCache)

sendMultiAction

void sendMultiAction(Map<ServerName,MultiAction> actionsByServer,
 int numAttempt,
 List<Action> actionsForReplicaThread,
 boolean reuseThread)

Send a multi action structure to the servers, after a delay depending on the attempt number. Asynchronous.

Parameters:

actionsByServer - the actions structured by regions

numAttempt - the attempt number.

actionsForReplicaThread - original actions for replica thread; null on non-first call.

getNewMultiActionRunnable

private Collection<? extends Runnable> getNewMultiActionRunnable(ServerName server,
 MultiAction multiAction,
 int numAttempt)

getBackoff

private Long getBackoff(ServerName server,
 byte[] regionName)

Parameters:

server - server location where the target region is hosted

regionName - name of the region which we are going to write some data

Returns:

the amount of time the client should wait until it submit a request to the specified server and region

startWaitingForReplicaCalls

private void startWaitingForReplicaCalls(List<Action> actionsForReplicaThread)

Starts waiting to issue replica calls on a different thread; or issues them immediately.

manageError

AsyncRequestFutureImpl.Retry manageError(int originalIndex,
 Row row,
 AsyncRequestFutureImpl.Retry canRetry,
 Throwable throwable,
 ServerName server)

Check that we can retry acts accordingly: logs, set the error status.

Parameters:

originalIndex - the position in the list sent

row - the row

canRetry - if false, we won't retry whatever the settings.

throwable - the throwable, if any (can be null)

server - the location, if any (can be null)

Returns:

true if the action can be retried, false otherwise.

failAll

private void failAll(MultiAction actions,
 ServerName server,
 int numAttempt,
 Throwable throwable)

Fail all the actions from this multiaction after an OperationTimeoutExceededException

Parameters:

actions - the actions still to do from the initial list

server - the destination

numAttempt - the number of attempts so far

throwable - the throwable that caused the failure

receiveGlobalFailure

private void receiveGlobalFailure(MultiAction rsActions,
 ServerName server,
 int numAttempt,
 Throwable t,
 boolean clearServerCache)

Resubmit all the actions from this multiaction after a failure.

Parameters:

rsActions - the actions still to do from the initial list

server - the destination

numAttempt - the number of attempts so far

t - the throwable (if any) that caused the resubmit

resubmit

private void resubmit(ServerName oldServer,
 List<Action> toReplay,
 int numAttempt,
 int failureCount,
 Throwable throwable)

Log as much info as possible, and, if there is something to replay, submit it again after a back off sleep.

logNoResubmit

private void logNoResubmit(ServerName oldServer,
 int numAttempt,
 int failureCount,
 Throwable throwable,
 int failed,
 int stopped)

receiveMultiAction

private void receiveMultiAction(MultiAction multiAction,
 ServerName server,
 MultiResponse responses,
 int numAttempt)

Called when we receive the result of a server query.

Parameters:

multiAction - - the multiAction we sent

server - - the location. It's used as a server name.

responses - - the response, if any

numAttempt - - the attempt

updateCachedLocations

private void updateCachedLocations(ServerName server,
 byte[] regionName,
 byte[] row,
 Throwable rowException)

invokeCallBack

private void invokeCallBack(byte[] regionName,
 byte[] row,
 CResult result)

cleanServerCache

private void cleanServerCache(ServerName server,
 Throwable regionException)

updateStats

protected void updateStats(ServerName server,
 MultiResponse resp)

createLog

private String createLog(int numAttempt,
 int failureCount,
 int replaySize,
 ServerName sn,
 Throwable error,
 long backOffTime,
 boolean willRetry,
 String startTime,
 int failed,
 int stopped)

setResult

private void setResult(Action action,
 Object result)

Sets the non-error result from a particular action.

Parameters:

action - Action (request) that the server responded to.

result - The result.

setError

private void setError(int index,
 Row row,
 Throwable throwable,
 ServerName server)

Sets the error from a particular action.

Parameters:

index - Original action index.

row - Original request.

throwable - The resulting error.

server - The source server.

isActionComplete

private boolean isActionComplete(int index,
 Row row)

Checks if the action is complete; used on error to prevent needless retries. Does not synchronize, assuming element index/field accesses are atomic. This is an opportunistic optimization check, doesn't have to be strict.

Parameters:

index - Original action index.

row - Original request.

trySetResultSimple

private AsyncRequestFutureImpl.ReplicaResultState trySetResultSimple(int index,
 Row row,
 boolean isError,
 Object result,
 ServerName server,
 boolean isFromReplica)

Tries to set the result or error for a particular action as if there were no replica calls.

Returns:

null if successful; replica state if there were in fact replica calls.

decActionCounter

private void decActionCounter(int index)

buildDetailedErrorMsg

private String buildDetailedErrorMsg(String string,
 int index)

waitUntilDone

public void waitUntilDone()
                   throws InterruptedIOException

Description copied from interface: AsyncRequestFuture

Wait until all tasks are executed, successfully or not.

Specified by:

waitUntilDone in interface AsyncRequestFuture

Throws:

InterruptedIOException

waitUntilDone

private boolean waitUntilDone(long cutoff)
                       throws InterruptedException

Throws:

InterruptedException

hasError

public boolean hasError()

Specified by:

hasError in interface AsyncRequestFuture

getFailedOperations

public List<? extends Row> getFailedOperations()

Specified by:

getFailedOperations in interface AsyncRequestFuture

getErrors

public RetriesExhaustedWithDetailsException getErrors()

Specified by:

getErrors in interface AsyncRequestFuture

getResults

public Object[] getResults()
                    throws InterruptedIOException

Specified by:

getResults in interface AsyncRequestFuture

Throws:

InterruptedIOException

createServerErrorTracker

private 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

createCallable

private MultiServerCallable createCallable(ServerName server,
 TableName tableName,
 MultiAction multi)

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

updateResult

private void updateResult(int index,
 Object result)

getNumberOfActionsInProgress

long getNumberOfActionsInProgress()