HBASE-16631 Extract AsyncRequestFuture related code from AsyncProcess

Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/2cf8907d
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/2cf8907d
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/2cf8907d

Branch: refs/heads/master
Commit: 2cf8907db53b84a0118acc1edd1dfb9b37abe8b7
Parents: b6b7236
Author: chenheng <chenh...@apache.org>
Authored: Sat Sep 17 00:35:23 2016 +0800
Committer: chenheng <chenh...@apache.org>
Committed: Sat Sep 17 00:35:23 2016 +0800

----------------------------------------------------------------------
 .../hadoop/hbase/client/AsyncProcess.java       | 1375 +-----------------
 .../hadoop/hbase/client/AsyncRequestFuture.java |   40 +
 .../hbase/client/AsyncRequestFutureImpl.java    | 1290 ++++++++++++++++
 .../apache/hadoop/hbase/client/BatchErrors.java |   69 +
 .../org/apache/hadoop/hbase/client/HTable.java  |    1 -
 .../hadoop/hbase/client/HTableMultiplexer.java  |    1 -
 .../hadoop/hbase/client/TestAsyncProcess.java   |   48 +-
 .../hadoop/hbase/client/TestReplicasClient.java |    2 -
 8 files changed, 1481 insertions(+), 1345 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/2cf8907d/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
----------------------------------------------------------------------
diff --git 
a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java 
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
index 93b17bc..2ffb2e3 100644
--- 
a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
+++ 
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
@@ -20,52 +20,41 @@
 package org.apache.hadoop.hbase.client;
 
 import com.google.common.annotations.VisibleForTesting;
-import java.io.InterruptedIOException;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-import org.apache.hadoop.hbase.client.AsyncProcess.RowChecker.ReturnCode;
-import org.apache.hadoop.hbase.client.backoff.ServerStatistics;
-import org.apache.hadoop.hbase.client.coprocessor.Batch;
-import org.apache.hadoop.hbase.DoNotRetryIOException;
-import org.apache.hadoop.hbase.exceptions.ClientExceptionsUtil;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.HRegionLocation;
-import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
-import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
-import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.RegionLocations;
-import org.apache.hadoop.hbase.RetryImmediatelyException;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.AsyncProcess.RowChecker.ReturnCode;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
+import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdge;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.apache.htrace.Trace;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * This class  allows a continuous flow of requests. It's written to be 
compatible with a
@@ -124,7 +113,7 @@ class AsyncProcess {
    */
   public static final String LOG_DETAILS_FOR_BATCH_ERROR = 
"hbase.client.log.batcherrors.details";
 
-  private final int thresholdToLogUndoneTaskDetails;
+  protected final int thresholdToLogUndoneTaskDetails;
   private static final String THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS =
       "hbase.client.threshold.log.details";
   private static final int DEFAULT_THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS = 10;
@@ -150,21 +139,6 @@ class AsyncProcess {
   public static final long DEFAULT_HBASE_CLIENT_MAX_SUBMIT_HEAPSIZE = 
DEFAULT_HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE;
 
   /**
-   * The context used to wait for results from one submit call.
-   * 1) If AsyncProcess is set to track errors globally, and not per call (for 
HTable puts),
-   *    then errors and failed operations in this object will reflect global 
errors.
-   * 2) If submit call is made with needResults false, results will not be 
saved.
-   *  */
-  public static interface AsyncRequestFuture {
-    public boolean hasError();
-    public RetriesExhaustedWithDetailsException getErrors();
-    public List<? extends Row> getFailedOperations();
-    public Object[] getResults() throws InterruptedIOException;
-    /** Wait until all tasks are executed, successfully or not. */
-    public void waitUntilDone() throws InterruptedIOException;
-  }
-
-  /**
    * Return value from a submit that didn't contain any requests.
    */
   private static final AsyncRequestFuture NO_REQS_RESULT = new 
AsyncRequestFuture() {
@@ -195,28 +169,6 @@ class AsyncProcess {
     }
   };
 
-  /** Sync point for calls to multiple replicas for the same user request 
(Get).
-   * Created and put in the results array (we assume replica calls require 
results) when
-   * the replica calls are launched. See results for details of this process.
-   * POJO, all fields are public. To modify them, the object itself is locked. 
*/
-  private static class ReplicaResultState {
-    public ReplicaResultState(int callCount) {
-      this.callCount = callCount;
-    }
-
-    /** Number of calls outstanding, or 0 if a call succeeded (even with 
others outstanding). */
-    int callCount;
-    /** Errors for which it is not decided whether we will report them to 
user. If one of the
-     * calls succeeds, we will discard the errors that may have happened in 
the other calls. */
-    BatchErrors replicaErrors = null;
-
-    @Override
-    public String toString() {
-      return "[call count " + callCount + "; errors " + replicaErrors + "]";
-    }
-  }
-
-
   // TODO: many of the fields should be made private
   protected final long id;
 
@@ -232,7 +184,7 @@ class AsyncProcess {
   protected final ConcurrentMap<ServerName, AtomicInteger> 
taskCounterPerServer =
       new ConcurrentHashMap<ServerName, AtomicInteger>();
   // Start configuration settings.
-  private final int startLogErrorsCnt;
+  protected final int startLogErrorsCnt;
 
   /**
    * The number of tasks simultaneously executed on the cluster.
@@ -262,49 +214,9 @@ class AsyncProcess {
   protected int timeout;
   protected long primaryCallTimeoutMicroseconds;
   /** Whether to log details for batch errors */
-  private final boolean logBatchErrorDetails;
+  protected final boolean logBatchErrorDetails;
   // End configuration settings.
 
-  protected static class BatchErrors {
-    private final List<Throwable> throwables = new ArrayList<Throwable>();
-    private final List<Row> actions = new ArrayList<Row>();
-    private final List<String> addresses = new ArrayList<String>();
-
-    public synchronized void add(Throwable ex, Row row, ServerName serverName) 
{
-      if (row == null){
-        throw new IllegalArgumentException("row cannot be null. location=" + 
serverName);
-      }
-
-      throwables.add(ex);
-      actions.add(row);
-      addresses.add(serverName != null ? serverName.toString() : "null");
-    }
-
-    public boolean hasErrors() {
-      return !throwables.isEmpty();
-    }
-
-    private synchronized RetriesExhaustedWithDetailsException 
makeException(boolean logDetails) {
-      if (logDetails) {
-        LOG.error("Exception occurred! Exception details: " + throwables + 
";\nActions: "
-            + actions);
-      }
-      return new RetriesExhaustedWithDetailsException(new 
ArrayList<Throwable>(throwables),
-          new ArrayList<Row>(actions), new ArrayList<String>(addresses));
-    }
-
-    public synchronized void clear() {
-      throwables.clear();
-      actions.clear();
-      addresses.clear();
-    }
-
-    public synchronized void merge(BatchErrors other) {
-      throwables.addAll(other.throwables);
-      actions.addAll(other.actions);
-      addresses.addAll(other.addresses);
-    }
-  }
   public AsyncProcess(ClusterConnection hc, Configuration conf, 
ExecutorService pool,
       RpcRetryingCallerFactory rpcCaller, boolean useGlobalErrors,
       RpcControllerFactory rpcFactory, int rpcTimeout) {
@@ -383,7 +295,7 @@ class AsyncProcess {
    * @return pool if non null, otherwise returns this.pool if non null, 
otherwise throws
    *         RuntimeException
    */
-  private ExecutorService getPool(ExecutorService pool) {
+  protected ExecutorService getPool(ExecutorService pool) {
     if (pool != null) {
       return pool;
     }
@@ -528,7 +440,7 @@ class AsyncProcess {
         int originalIndex = locationErrorRows.get(i);
         Row row = retainedActions.get(originalIndex).getAction();
         ars.manageError(originalIndex, row,
-            Retry.NO_LOCATION_PROBLEM, locationErrors.get(i), null);
+            AsyncRequestFutureImpl.Retry.NO_LOCATION_PROBLEM, 
locationErrors.get(i), null);
       }
     }
     ars.sendMultiAction(actionsByServer, 1, null, false);
@@ -538,12 +450,13 @@ class AsyncProcess {
   /**
    * Helper that is used when grouping the actions per region server.
    *
-   * @param loc - the destination. Must not be null.
+   * @param server - server
+   * @param regionName - regionName
    * @param action - the action to add to the multiaction
    * @param actionsByServer the multiaction per server
    * @param nonceGroup Nonce group.
    */
-  private static void addAction(ServerName server, byte[] regionName, 
Action<Row> action,
+  static void addAction(ServerName server, byte[] regionName, Action<Row> 
action,
       Map<ServerName, MultiAction<Row>> actionsByServer, long nonceGroup) {
     MultiAction<Row> multiAction = actionsByServer.get(server);
     if (multiAction == null) {
@@ -598,1210 +511,22 @@ class AsyncProcess {
     return ars;
   }
 
-  private static void setNonce(NonceGenerator ng, Row r, Action<Row> action) {
+  private void setNonce(NonceGenerator ng, Row r, Action<Row> action) {
     if (!(r instanceof Append) && !(r instanceof Increment)) return;
     action.setNonce(ng.newNonce()); // Action handles NO_NONCE, so it's ok if 
ng is disabled.
   }
 
-  /**
-   * 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.
-   */
-  protected class AsyncRequestFutureImpl<CResult> implements 
AsyncRequestFuture {
-
-    /**
-     * 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).
-     * This is done on a separate thread because we don't want to wait on user 
thread for
-     * our asynchronous call, and usually we have to wait before making 
replica calls.
-     */
-    private final class ReplicaCallIssuingRunnable implements Runnable {
-      private final long startTime;
-      private final List<Action<Row>> initialActions;
-
-      public ReplicaCallIssuingRunnable(List<Action<Row>> initialActions, long 
startTime) {
-        this.initialActions = initialActions;
-        this.startTime = startTime;
-      }
-
-      @Override
-      public void run() {
-        boolean done = false;
-        if (primaryCallTimeoutMicroseconds > 0) {
-          try {
-            done = waitUntilDone(startTime * 1000L + 
primaryCallTimeoutMicroseconds);
-          } catch (InterruptedException ex) {
-            LOG.error("Replica thread was interrupted - no replica calls: " + 
ex.getMessage());
-            return;
-          }
-        }
-        if (done) return; // Done within primary timeout
-        Map<ServerName, MultiAction<Row>> actionsByServer =
-            new HashMap<ServerName, MultiAction<Row>>();
-        List<Action<Row>> unknownLocActions = new ArrayList<Action<Row>>();
-        if (replicaGetIndices == null) {
-          for (int i = 0; i < results.length; ++i) {
-            addReplicaActions(i, actionsByServer, unknownLocActions);
-          }
-        } else {
-          for (int replicaGetIndice : replicaGetIndices) {
-            addReplicaActions(replicaGetIndice, actionsByServer, 
unknownLocActions);
-          }
-        }
-        if (!actionsByServer.isEmpty()) {
-          sendMultiAction(actionsByServer, 1, null, 
unknownLocActions.isEmpty());
-        }
-        if (!unknownLocActions.isEmpty()) {
-          actionsByServer = new HashMap<ServerName, MultiAction<Row>>();
-          for (Action<Row> action : unknownLocActions) {
-            addReplicaActionsAgain(action, actionsByServer);
-          }
-          // Some actions may have completely failed, they are handled inside 
addAgain.
-          if (!actionsByServer.isEmpty()) {
-            sendMultiAction(actionsByServer, 1, null, true);
-          }
-        }
-      }
-
-      /**
-       * Add replica actions to action map by server.
-       * @param index Index of the original action.
-       * @param actionsByServer The map by server to add it to.
-       */
-      private void addReplicaActions(int index, Map<ServerName, 
MultiAction<Row>> actionsByServer,
-          List<Action<Row>> unknownReplicaActions) {
-        if (results[index] != null) return; // opportunistic. Never goes from 
non-null to null.
-        Action<Row> action = initialActions.get(index);
-        RegionLocations loc = findAllLocationsOrFail(action, true);
-        if (loc == null) return;
-        HRegionLocation[] locs = loc.getRegionLocations();
-        if (locs.length == 1) {
-          LOG.warn("No replicas found for " + action.getAction());
-          return;
-        }
-        synchronized (replicaResultLock) {
-          // Don't run replica calls if the original has finished. We could do 
it e.g. if
-          // original has already failed before first replica call (unlikely 
given retries),
-          // but that would require additional synchronization w.r.t. 
returning to caller.
-          if (results[index] != null) return;
-          // We set the number of calls here. After that any path must call 
setResult/setError.
-          // True even for replicas that are not found - if we refuse to send 
we MUST set error.
-          results[index] = new ReplicaResultState(locs.length);
-        }
-        for (int i = 1; i < locs.length; ++i) {
-          Action<Row> replicaAction = new Action<Row>(action, i);
-          if (locs[i] != null) {
-            addAction(locs[i].getServerName(), 
locs[i].getRegionInfo().getRegionName(),
-                replicaAction, actionsByServer, nonceGroup);
-          } else {
-            unknownReplicaActions.add(replicaAction);
-          }
-        }
-      }
-
-      private void addReplicaActionsAgain(
-          Action<Row> action, Map<ServerName, MultiAction<Row>> 
actionsByServer) {
-        if (action.getReplicaId() == RegionReplicaUtil.DEFAULT_REPLICA_ID) {
-          throw new AssertionError("Cannot have default replica here");
-        }
-        HRegionLocation loc = getReplicaLocationOrFail(action);
-        if (loc == null) return;
-        addAction(loc.getServerName(), loc.getRegionInfo().getRegionName(),
-            action, actionsByServer, nonceGroup);
-      }
-    }
-
-    /**
-     * Runnable (that can be submitted to thread pool) that submits 
MultiAction to a
-     * single server. The server call is synchronous, therefore we do it on a 
thread pool.
-     */
-    private final class SingleServerRequestRunnable implements Runnable {
-      private final MultiAction<Row> multiAction;
-      private final int numAttempt;
-      private final ServerName server;
-      private final Set<CancellableRegionServerCallable> callsInProgress;
-      private Long heapSize = null;
-      private SingleServerRequestRunnable(
-          MultiAction<Row> multiAction, int numAttempt, ServerName server,
-          Set<CancellableRegionServerCallable> callsInProgress) {
-        this.multiAction = multiAction;
-        this.numAttempt = numAttempt;
-        this.server = server;
-        this.callsInProgress = callsInProgress;
-      }
-
-      @VisibleForTesting
-      long heapSize() {
-        if (heapSize != null) {
-          return heapSize;
-        }
-        heapSize = 0L;
-        for (Map.Entry<byte[], List<Action<Row>>> e: 
this.multiAction.actions.entrySet()) {
-          List<Action<Row>> actions = e.getValue();
-          for (Action<Row> action: actions) {
-            Row row = action.getAction();
-            if (row instanceof Mutation) {
-              heapSize += ((Mutation) row).heapSize();
-            }
-          }
-        }
-        return heapSize;
-      }
-
-      @Override
-      public void run() {
-        AbstractResponse res = null;
-        CancellableRegionServerCallable callable = currentCallable;
-        try {
-          // setup the callable based on the actions, if we don't have one 
already from the request
-          if (callable == null) {
-            callable = createCallable(server, tableName, multiAction);
-          }
-          RpcRetryingCaller<AbstractResponse> caller = createCaller(callable);
-          try {
-            if (callsInProgress != null) {
-              callsInProgress.add(callable);
-            }
-            res = caller.callWithoutRetries(callable, currentCallTotalTimeout);
-            if (res == null) {
-              // Cancelled
-              return;
-            }
-          } catch (IOException e) {
-            // The service itself failed . It may be an error coming from the 
communication
-            //   layer, but, as well, a functional error raised by the server.
-            receiveGlobalFailure(multiAction, server, numAttempt, e);
-            return;
-          } catch (Throwable t) {
-            // This should not happen. Let's log & retry anyway.
-            LOG.error("#" + id + ", Caught throwable while calling. This is 
unexpected." +
-                " Retrying. Server is " + server + ", tableName=" + tableName, 
t);
-            receiveGlobalFailure(multiAction, server, numAttempt, t);
-            return;
-          }
-          if (res.type() == AbstractResponse.ResponseType.MULTI) {
-            // Normal case: we received an answer from the server, and it's 
not an exception.
-            receiveMultiAction(multiAction, server, (MultiResponse) res, 
numAttempt);
-          } else {
-            if (results != null) {
-              SingleResponse singleResponse = (SingleResponse) res;
-              results[0] = singleResponse.getEntry();
-            }
-            decActionCounter(1);
-          }
-        } catch (Throwable t) {
-          // Something really bad happened. We are on the send thread that 
will now die.
-          LOG.error("Internal AsyncProcess #" + id + " error for "
-              + tableName + " processing for " + server, t);
-          throw new RuntimeException(t);
-        } finally {
-          decTaskCounters(multiAction.getRegions(), server);
-          if (callsInProgress != null && callable != null && res != null) {
-            callsInProgress.remove(callable);
-          }
-        }
-      }
-    }
-
-    private final Batch.Callback<CResult> callback;
-    private final BatchErrors errors;
-    private final ConnectionImplementation.ServerErrorTracker errorsByServer;
-    private final ExecutorService pool;
-    private final Set<CancellableRegionServerCallable> callsInProgress;
-
-
-    private final TableName tableName;
-    private final AtomicLong actionsInProgress = new AtomicLong(-1);
-    /**
-     * 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.
-     */
-    private final Object replicaResultLock = new Object();
-    /**
-     * 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.
-     */
-    private final Object[] results;
-    /**
-     * Indices of replica gets in results. If null, all or no actions are 
replica-gets.
-     */
-    private final int[] replicaGetIndices;
-    private final boolean hasAnyReplicaGets;
-    private final long nonceGroup;
-    private CancellableRegionServerCallable currentCallable;
-    private int currentCallTotalTimeout;
-    private final Map<ServerName, List<Long>> heapSizesByServer = new 
HashMap<>();
-    public AsyncRequestFutureImpl(TableName tableName, List<Action<Row>> 
actions, long nonceGroup,
-        ExecutorService pool, boolean needResults, Object[] results,
-        Batch.Callback<CResult> callback, CancellableRegionServerCallable 
callable, int timeout) {
-      this.pool = pool;
-      this.callback = callback;
-      this.nonceGroup = nonceGroup;
-      this.tableName = tableName;
-      this.actionsInProgress.set(actions.size());
-      if (results != null) {
-        assert needResults;
-        if (results.length != actions.size()) {
-          throw new AssertionError("results.length");
-        }
-        this.results = results;
-        for (int i = 0; i != this.results.length; ++i) {
-          results[i] = null;
-        }
-      } else {
-        this.results = needResults ? new Object[actions.size()] : null;
-      }
-      List<Integer> replicaGetIndices = null;
-      boolean hasAnyReplicaGets = false;
-      if (needResults) {
-        // Check to see if any requests might require replica calls.
-        // We expect that many requests will consist of all or no 
multi-replica gets; in such
-        // cases we would just use a boolean (hasAnyReplicaGets). If there's a 
mix, we will
-        // store the list of action indexes for which replica gets are 
possible, and set
-        // hasAnyReplicaGets to true.
-        boolean hasAnyNonReplicaReqs = false;
-        int posInList = 0;
-        for (Action<Row> action : actions) {
-          boolean isReplicaGet = isReplicaGet(action.getAction());
-          if (isReplicaGet) {
-            hasAnyReplicaGets = true;
-            if (hasAnyNonReplicaReqs) { // Mixed case
-              if (replicaGetIndices == null) {
-                replicaGetIndices = new ArrayList<Integer>(actions.size() - 1);
-              }
-              replicaGetIndices.add(posInList);
-            }
-          } else if (!hasAnyNonReplicaReqs) {
-            // The first non-multi-replica request in the action list.
-            hasAnyNonReplicaReqs = true;
-            if (posInList > 0) {
-              // Add all the previous requests to the index lists. We know 
they are all
-              // replica-gets because this is the first non-multi-replica 
request in the list.
-              replicaGetIndices = new ArrayList<Integer>(actions.size() - 1);
-              for (int i = 0; i < posInList; ++i) {
-                replicaGetIndices.add(i);
-              }
-            }
-          }
-          ++posInList;
-        }
-      }
-      this.hasAnyReplicaGets = hasAnyReplicaGets;
-      if (replicaGetIndices != null) {
-        this.replicaGetIndices = new int[replicaGetIndices.size()];
-        int i = 0;
-        for (Integer el : replicaGetIndices) {
-          this.replicaGetIndices[i++] = el;
-        }
-      } else {
-        this.replicaGetIndices = null;
-      }
-      this.callsInProgress = !hasAnyReplicaGets ? null :
-          Collections.newSetFromMap(
-              new ConcurrentHashMap<CancellableRegionServerCallable, 
Boolean>());
-
-      this.errorsByServer = createServerErrorTracker();
-      this.errors = (globalErrors != null) ? globalErrors : new BatchErrors();
-      this.currentCallable = callable;
-      this.currentCallTotalTimeout = timeout;
-    }
-
-    public Set<CancellableRegionServerCallable> getCallsInProgress() {
-      return callsInProgress;
-    }
-    @VisibleForTesting
-    Map<ServerName, List<Long>> getRequestHeapSize() {
-      return heapSizesByServer;
-    }
-
-    private SingleServerRequestRunnable 
addSingleServerRequestHeapSize(ServerName server,
-        SingleServerRequestRunnable runnable) {
-      List<Long> heapCount = heapSizesByServer.get(server);
-      if (heapCount == null) {
-        heapCount = new LinkedList<>();
-        heapSizesByServer.put(server, heapCount);
-      }
-      heapCount.add(runnable.heapSize());
-      return runnable;
-    }
-    /**
-     * Group a list of actions per region servers, and send them.
-     *
-     * @param currentActions - the list of row to submit
-     * @param numAttempt - the current numAttempt (first attempt is 1)
-     */
-    private void groupAndSendMultiAction(List<Action<Row>> currentActions, int 
numAttempt) {
-      Map<ServerName, MultiAction<Row>> actionsByServer =
-          new HashMap<ServerName, MultiAction<Row>>();
-
-      boolean isReplica = false;
-      List<Action<Row>> unknownReplicaActions = null;
-      for (Action<Row> action : currentActions) {
-        RegionLocations locs = findAllLocationsOrFail(action, true);
-        if (locs == null) continue;
-        boolean isReplicaAction = 
!RegionReplicaUtil.isDefaultReplica(action.getReplicaId());
-        if (isReplica && !isReplicaAction) {
-          // This is the property of the current implementation, not a 
requirement.
-          throw new AssertionError("Replica and non-replica actions in the 
same retry");
-        }
-        isReplica = isReplicaAction;
-        HRegionLocation loc = locs.getRegionLocation(action.getReplicaId());
-        if (loc == null || loc.getServerName() == null) {
-          if (isReplica) {
-            if (unknownReplicaActions == null) {
-              unknownReplicaActions = new ArrayList<Action<Row>>();
-            }
-            unknownReplicaActions.add(action);
-          } else {
-            // TODO: relies on primary location always being fetched
-            manageLocationError(action, null);
-          }
-        } else {
-          byte[] regionName = loc.getRegionInfo().getRegionName();
-          addAction(loc.getServerName(), regionName, action, actionsByServer, 
nonceGroup);
-        }
-      }
-      boolean doStartReplica = (numAttempt == 1 && !isReplica && 
hasAnyReplicaGets);
-      boolean hasUnknown = unknownReplicaActions != null && 
!unknownReplicaActions.isEmpty();
-
-      if (!actionsByServer.isEmpty()) {
-        // If this is a first attempt to group and send, no replicas, we need 
replica thread.
-        sendMultiAction(actionsByServer, numAttempt, (doStartReplica && 
!hasUnknown)
-            ? currentActions : null, numAttempt > 1 && !hasUnknown);
-      }
-
-      if (hasUnknown) {
-        actionsByServer = new HashMap<ServerName, MultiAction<Row>>();
-        for (Action<Row> action : unknownReplicaActions) {
-          HRegionLocation loc = getReplicaLocationOrFail(action);
-          if (loc == null) continue;
-          byte[] regionName = loc.getRegionInfo().getRegionName();
-          addAction(loc.getServerName(), regionName, action, actionsByServer, 
nonceGroup);
-        }
-        if (!actionsByServer.isEmpty()) {
-          sendMultiAction(
-              actionsByServer, numAttempt, doStartReplica ? currentActions : 
null, true);
-        }
-      }
-    }
-
-    private HRegionLocation getReplicaLocationOrFail(Action<Row> action) {
-      // We are going to try get location once again. For each action, we'll 
do it once
-      // from cache, because the previous calls in the loop might populate it.
-      int replicaId = action.getReplicaId();
-      RegionLocations locs = findAllLocationsOrFail(action, true);
-      if (locs == null) return null; // manageError already called
-      HRegionLocation loc = locs.getRegionLocation(replicaId);
-      if (loc == null || loc.getServerName() == null) {
-        locs = findAllLocationsOrFail(action, false);
-        if (locs == null) return null; // manageError already called
-        loc = locs.getRegionLocation(replicaId);
-      }
-      if (loc == null || loc.getServerName() == null) {
-        manageLocationError(action, null);
-        return null;
-      }
-      return loc;
-    }
-
-    private void manageLocationError(Action<Row> action, Exception ex) {
-      String msg = "Cannot get replica " + action.getReplicaId()
-          + " location for " + action.getAction();
-      LOG.error(msg);
-      if (ex == null) {
-        ex = new IOException(msg);
-      }
-      manageError(action.getOriginalIndex(), action.getAction(),
-          Retry.NO_LOCATION_PROBLEM, ex, null);
-    }
-
-    private RegionLocations findAllLocationsOrFail(Action<Row> action, boolean 
useCache) {
-      if (action.getAction() == null) throw new IllegalArgumentException("#" + 
id +
-          ", row cannot be null");
-      RegionLocations loc = null;
-      try {
-        loc = connection.locateRegion(
-            tableName, action.getAction().getRow(), useCache, true, 
action.getReplicaId());
-      } catch (IOException ex) {
-        manageLocationError(action, ex);
-      }
-      return loc;
-    }
-
-    /**
-     * Send a multi action structure to the servers, after a delay depending 
on the attempt
-     * number. Asynchronous.
-     *
-     * @param actionsByServer the actions structured by regions
-     * @param numAttempt the attempt number.
-     * @param actionsForReplicaThread original actions for replica thread; 
null on non-first call.
-     */
-    private void sendMultiAction(Map<ServerName, MultiAction<Row>> 
actionsByServer,
-        int numAttempt, List<Action<Row>> actionsForReplicaThread, boolean 
reuseThread) {
-      // Run the last item on the same thread if we are already on a send 
thread.
-      // We hope most of the time it will be the only item, so we can cut down 
on threads.
-      int actionsRemaining = actionsByServer.size();
-      // This iteration is by server (the HRegionLocation comparator is by 
server portion only).
-      for (Map.Entry<ServerName, MultiAction<Row>> e : 
actionsByServer.entrySet()) {
-        ServerName server = e.getKey();
-        MultiAction<Row> multiAction = e.getValue();
-        Collection<? extends Runnable> runnables = 
getNewMultiActionRunnable(server, multiAction,
-            numAttempt);
-        // make sure we correctly count the number of runnables before we try 
to reuse the send
-        // thread, in case we had to split the request into different 
runnables because of backoff
-        if (runnables.size() > actionsRemaining) {
-          actionsRemaining = runnables.size();
-        }
-
-        // run all the runnables
-        for (Runnable runnable : runnables) {
-          if ((--actionsRemaining == 0) && reuseThread) {
-            runnable.run();
-          } else {
-            try {
-              pool.submit(runnable);
-            } catch (Throwable t) {
-              if (t instanceof RejectedExecutionException) {
-                // This should never happen. But as the pool is provided by 
the end user,
-               // let's secure this a little.
-               LOG.warn("#" + id + ", the task was rejected by the pool. This 
is unexpected." +
-                  " Server is " + server.getServerName(), t);
-              } else {
-                // see #HBASE-14359 for more details
-                LOG.warn("Caught unexpected exception/error: ", t);
-              }
-              decTaskCounters(multiAction.getRegions(), server);
-              // We're likely to fail again, but this will increment the 
attempt counter,
-             // so it will finish.
-              receiveGlobalFailure(multiAction, server, numAttempt, t);
-            }
-          }
-        }
-      }
-
-      if (actionsForReplicaThread != null) {
-        startWaitingForReplicaCalls(actionsForReplicaThread);
-      }
-    }
-
-    private Collection<? extends Runnable> 
getNewMultiActionRunnable(ServerName server,
-        MultiAction<Row> multiAction,
-        int numAttempt) {
-      // no stats to manage, just do the standard action
-      if (AsyncProcess.this.connection.getStatisticsTracker() == null) {
-        if (connection.getConnectionMetrics() != null) {
-          connection.getConnectionMetrics().incrNormalRunners();
-        }
-        incTaskCounters(multiAction.getRegions(), server);
-        SingleServerRequestRunnable runnable = 
addSingleServerRequestHeapSize(server,
-          new SingleServerRequestRunnable(multiAction, numAttempt, server, 
callsInProgress));
-        return 
Collections.singletonList(Trace.wrap("AsyncProcess.sendMultiAction", runnable));
-      }
-
-      // group the actions by the amount of delay
-      Map<Long, DelayingRunner> actions = new HashMap<Long, 
DelayingRunner>(multiAction
-          .size());
-
-      // split up the actions
-      for (Map.Entry<byte[], List<Action<Row>>> e : 
multiAction.actions.entrySet()) {
-        Long backoff = getBackoff(server, e.getKey());
-        DelayingRunner runner = actions.get(backoff);
-        if (runner == null) {
-          actions.put(backoff, new DelayingRunner(backoff, e));
-        } else {
-          runner.add(e);
-        }
-      }
-
-      List<Runnable> toReturn = new ArrayList<Runnable>(actions.size());
-      for (DelayingRunner runner : actions.values()) {
-        incTaskCounters(runner.getActions().getRegions(), server);
-        String traceText = "AsyncProcess.sendMultiAction";
-        Runnable runnable = addSingleServerRequestHeapSize(server,
-          new SingleServerRequestRunnable(runner.getActions(), numAttempt, 
server, callsInProgress));
-        // use a delay runner only if we need to sleep for some time
-        if (runner.getSleepTime() > 0) {
-          runner.setRunner(runnable);
-          traceText = "AsyncProcess.clientBackoff.sendMultiAction";
-          runnable = runner;
-          if (connection.getConnectionMetrics() != null) {
-            connection.getConnectionMetrics().incrDelayRunners();
-            
connection.getConnectionMetrics().updateDelayInterval(runner.getSleepTime());
-          }
-        } else {
-          if (connection.getConnectionMetrics() != null) {
-            connection.getConnectionMetrics().incrNormalRunners();
-          }
-        }
-        runnable = Trace.wrap(traceText, runnable);
-        toReturn.add(runnable);
-
-      }
-      return toReturn;
-    }
-
-    /**
-     * @param server server location where the target region is hosted
-     * @param regionName name of the region which we are going to write some 
data
-     * @return the amount of time the client should wait until it submit a 
request to the
-     * specified server and region
-     */
-    private Long getBackoff(ServerName server, byte[] regionName) {
-      ServerStatisticTracker tracker = 
AsyncProcess.this.connection.getStatisticsTracker();
-      ServerStatistics stats = tracker.getStats(server);
-      return AsyncProcess.this.connection.getBackoffPolicy()
-          .getBackoffTime(server, regionName, stats);
-    }
-
-    /**
-     * Starts waiting to issue replica calls on a different thread; or issues 
them immediately.
-     */
-    private void startWaitingForReplicaCalls(List<Action<Row>> 
actionsForReplicaThread) {
-      long startTime = EnvironmentEdgeManager.currentTime();
-      ReplicaCallIssuingRunnable replicaRunnable = new 
ReplicaCallIssuingRunnable(
-          actionsForReplicaThread, startTime);
-      if (primaryCallTimeoutMicroseconds == 0) {
-        // Start replica calls immediately.
-        replicaRunnable.run();
-      } else {
-        // Start the thread that may kick off replica gets.
-        // TODO: we could do it on the same thread, but it's a user thread, 
might be a bad idea.
-        try {
-          pool.submit(replicaRunnable);
-        } catch (RejectedExecutionException ree) {
-          LOG.warn("#" + id + ", replica task was rejected by the pool - no 
replica calls", ree);
-        }
-      }
-    }
-
-    /**
-     * Check that we can retry acts accordingly: logs, set the error status.
-     *
-     * @param originalIndex the position in the list sent
-     * @param row           the row
-     * @param canRetry      if false, we won't retry whatever the settings.
-     * @param throwable     the throwable, if any (can be null)
-     * @param server        the location, if any (can be null)
-     * @return true if the action can be retried, false otherwise.
-     */
-    public Retry manageError(int originalIndex, Row row, Retry canRetry,
-                                Throwable throwable, ServerName server) {
-      if (canRetry == Retry.YES
-          && throwable != null && throwable instanceof DoNotRetryIOException) {
-        canRetry = Retry.NO_NOT_RETRIABLE;
-      }
-
-      if (canRetry != Retry.YES) {
-        // Batch.Callback<Res> was not called on failure in 0.94. We keep this.
-        setError(originalIndex, row, throwable, server);
-      } else if (isActionComplete(originalIndex, row)) {
-        canRetry = Retry.NO_OTHER_SUCCEEDED;
-      }
-      return canRetry;
-    }
-
-    /**
-     * Resubmit all the actions from this multiaction after a failure.
-     *
-     * @param rsActions  the actions still to do from the initial list
-     * @param server   the destination
-     * @param numAttempt the number of attempts so far
-     * @param t the throwable (if any) that caused the resubmit
-     */
-    private void receiveGlobalFailure(
-        MultiAction<Row> rsActions, ServerName server, int numAttempt, 
Throwable t) {
-      errorsByServer.reportServerError(server);
-      Retry canRetry = errorsByServer.canTryMore(numAttempt)
-          ? Retry.YES : Retry.NO_RETRIES_EXHAUSTED;
-
-      if (tableName == null && 
ClientExceptionsUtil.isMetaClearingException(t)) {
-        // tableName is null when we made a cross-table RPC call.
-        connection.clearCaches(server);
-      }
-      int failed = 0, stopped = 0;
-      List<Action<Row>> toReplay = new ArrayList<Action<Row>>();
-      for (Map.Entry<byte[], List<Action<Row>>> e : 
rsActions.actions.entrySet()) {
-        byte[] regionName = e.getKey();
-        byte[] row = e.getValue().iterator().next().getAction().getRow();
-        // Do not use the exception for updating cache because it might be 
coming from
-        // any of the regions in the MultiAction.
-        try {
-          if (tableName != null) {
-            connection.updateCachedLocations(tableName, regionName, row,
-              ClientExceptionsUtil.isMetaClearingException(t) ? null : t, 
server);
-          }
-        } catch (Throwable ex) {
-          // That should never happen, but if it did, we want to make sure
-          // we still process errors
-          LOG.error("Couldn't update cached region locations: " + ex);
-        }
-        for (Action<Row> action : e.getValue()) {
-          Retry retry = manageError(
-              action.getOriginalIndex(), action.getAction(), canRetry, t, 
server);
-          if (retry == Retry.YES) {
-            toReplay.add(action);
-          } else if (retry == Retry.NO_OTHER_SUCCEEDED) {
-            ++stopped;
-          } else {
-            ++failed;
-          }
-        }
-      }
-
-      if (toReplay.isEmpty()) {
-        logNoResubmit(server, numAttempt, rsActions.size(), t, failed, 
stopped);
-      } else {
-        resubmit(server, toReplay, numAttempt, rsActions.size(), t);
-      }
-    }
-
-    /**
-     * Log as much info as possible, and, if there is something to replay,
-     * submit it again after a back off sleep.
-     */
-    private void resubmit(ServerName oldServer, List<Action<Row>> toReplay,
-        int numAttempt, int failureCount, Throwable throwable) {
-      // We have something to replay. We're going to sleep a little before.
-
-      // We have two contradicting needs here:
-      //  1) We want to get the new location after having slept, as it may 
change.
-      //  2) We want to take into account the location when calculating the 
sleep time.
-      //  3) If all this is just because the response needed to be chunked try 
again FAST.
-      // It should be possible to have some heuristics to take the right 
decision. Short term,
-      //  we go for one.
-      boolean retryImmediately = throwable instanceof 
RetryImmediatelyException;
-      int nextAttemptNumber = retryImmediately ? numAttempt : numAttempt + 1;
-      long backOffTime = retryImmediately ? 0 :
-          errorsByServer.calculateBackoffTime(oldServer, pause);
-      if (numAttempt > startLogErrorsCnt) {
-        // We use this value to have some logs when we have multiple failures, 
but not too many
-        //  logs, as errors are to be expected when a region moves, splits and 
so on
-        LOG.info(createLog(numAttempt, failureCount, toReplay.size(),
-            oldServer, throwable, backOffTime, true, null, -1, -1));
-      }
-
-      try {
-        if (backOffTime > 0) {
-          Thread.sleep(backOffTime);
-        }
-      } catch (InterruptedException e) {
-        LOG.warn("#" + id + ", not sent: " + toReplay.size() + " operations, " 
+ oldServer, e);
-        Thread.currentThread().interrupt();
-        return;
-      }
-
-      groupAndSendMultiAction(toReplay, nextAttemptNumber);
-    }
-
-    private void logNoResubmit(ServerName oldServer, int numAttempt,
-        int failureCount, Throwable throwable, int failed, int stopped) {
-      if (failureCount != 0 || numAttempt > startLogErrorsCnt + 1) {
-        String timeStr = new 
Date(errorsByServer.getStartTrackingTime()).toString();
-        String logMessage = createLog(numAttempt, failureCount, 0, oldServer,
-            throwable, -1, false, timeStr, failed, stopped);
-        if (failed != 0) {
-          // Only log final failures as warning
-          LOG.warn(logMessage);
-        } else {
-          LOG.info(logMessage);
-        }
-      }
-    }
-
-    /**
-     * Called when we receive the result of a server query.
-     *
-     * @param multiAction    - the multiAction we sent
-     * @param server       - the location. It's used as a server name.
-     * @param responses      - the response, if any
-     * @param numAttempt     - the attempt
-     */
-    private void receiveMultiAction(MultiAction<Row> multiAction,
-        ServerName server, MultiResponse responses, int numAttempt) {
-       assert responses != null;
-
-      // Success or partial success
-      // Analyze detailed results. We can still have individual failures to be 
redo.
-      // two specific throwables are managed:
-      //  - DoNotRetryIOException: we continue to retry for other actions
-      //  - RegionMovedException: we update the cache with the new region 
location
-
-      List<Action<Row>> toReplay = new ArrayList<Action<Row>>();
-      Throwable throwable = null;
-      int failureCount = 0;
-      boolean canRetry = true;
-
-      Map<byte[], MultiResponse.RegionResult> results = responses.getResults();
-      updateStats(server, results);
-
-      int failed = 0, stopped = 0;
-      // Go by original action.
-      for (Map.Entry<byte[], List<Action<Row>>> regionEntry : 
multiAction.actions.entrySet()) {
-        byte[] regionName = regionEntry.getKey();
-        Map<Integer, Object> regionResults = results.get(regionName) == null
-            ?  null : results.get(regionName).result;
-        if (regionResults == null) {
-          if (!responses.getExceptions().containsKey(regionName)) {
-            LOG.error("Server sent us neither results nor exceptions for "
-                + Bytes.toStringBinary(regionName));
-            responses.getExceptions().put(regionName, new 
RuntimeException("Invalid response"));
-          }
-          continue;
-        }
-        boolean regionFailureRegistered = false;
-        for (Action<Row> sentAction : regionEntry.getValue()) {
-          Object result = regionResults.get(sentAction.getOriginalIndex());
-          // Failure: retry if it's make sense else update the errors lists
-          if (result == null || result instanceof Throwable) {
-            Row row = sentAction.getAction();
-            throwable = ClientExceptionsUtil.findException(result);
-            // Register corresponding failures once per server/once per region.
-            if (!regionFailureRegistered) {
-              regionFailureRegistered = true;
-              try {
-                connection.updateCachedLocations(
-                  tableName, regionName, row.getRow(), result, server);
-              } catch (Throwable ex) {
-                // That should never happen, but if it did, we want to make 
sure
-                // we still process errors
-                LOG.error("Couldn't update cached region locations: " + ex);
-              }
-            }
-            if (failureCount == 0) {
-              errorsByServer.reportServerError(server);
-              // We determine canRetry only once for all calls, after 
reporting server failure.
-              canRetry = errorsByServer.canTryMore(numAttempt);
-            }
-            ++failureCount;
-            Retry retry = manageError(sentAction.getOriginalIndex(), row,
-                canRetry ? Retry.YES : Retry.NO_RETRIES_EXHAUSTED, (Throwable) 
result, server);
-            if (retry == Retry.YES) {
-              toReplay.add(sentAction);
-            } else if (retry == Retry.NO_OTHER_SUCCEEDED) {
-              ++stopped;
-            } else {
-              ++failed;
-            }
-          } else {
-            if (callback != null) {
-              try {
-                //noinspection unchecked
-                // TODO: would callback expect a replica region name if it 
gets one?
-                this.callback.update(regionName, 
sentAction.getAction().getRow(), (CResult) result);
-              } catch (Throwable t) {
-                LOG.error("User callback threw an exception for "
-                    + Bytes.toStringBinary(regionName) + ", ignoring", t);
-              }
-            }
-            setResult(sentAction, result);
-          }
-        }
-      }
-
-      // The failures global to a region. We will use for multiAction we sent 
previously to find the
-      //   actions to replay.
-      for (Map.Entry<byte[], Throwable> throwableEntry : 
responses.getExceptions().entrySet()) {
-        throwable = throwableEntry.getValue();
-        byte[] region = throwableEntry.getKey();
-        List<Action<Row>> actions = multiAction.actions.get(region);
-        if (actions == null || actions.isEmpty()) {
-          throw new IllegalStateException("Wrong response for the region: " +
-              HRegionInfo.encodeRegionName(region));
-        }
-
-        if (failureCount == 0) {
-          errorsByServer.reportServerError(server);
-          canRetry = errorsByServer.canTryMore(numAttempt);
-        }
-        if (null == tableName && 
ClientExceptionsUtil.isMetaClearingException(throwable)) {
-          // For multi-actions, we don't have a table name, but we want to 
make sure to clear the
-          // cache in case there were location-related exceptions. We don't to 
clear the cache
-          // for every possible exception that comes through, however.
-          connection.clearCaches(server);
-        } else {
-          try {
-            connection.updateCachedLocations(
-              tableName, region, actions.get(0).getAction().getRow(), 
throwable, server);
-          } catch (Throwable ex) {
-            // That should never happen, but if it did, we want to make sure
-            // we still process errors
-            LOG.error("Couldn't update cached region locations: " + ex);
-          }
-        }
-        failureCount += actions.size();
-
-        for (Action<Row> action : actions) {
-          Row row = action.getAction();
-          Retry retry = manageError(action.getOriginalIndex(), row,
-              canRetry ? Retry.YES : Retry.NO_RETRIES_EXHAUSTED, throwable, 
server);
-          if (retry == Retry.YES) {
-            toReplay.add(action);
-          } else if (retry == Retry.NO_OTHER_SUCCEEDED) {
-            ++stopped;
-          } else {
-            ++failed;
-          }
-        }
-      }
-      if (toReplay.isEmpty()) {
-        logNoResubmit(server, numAttempt, failureCount, throwable, failed, 
stopped);
-      } else {
-        resubmit(server, toReplay, numAttempt, failureCount, throwable);
-      }
-    }
-
-    private String createLog(int numAttempt, int failureCount, int replaySize, 
ServerName sn,
-        Throwable error, long backOffTime, boolean willRetry, String startTime,
-        int failed, int stopped) {
-      StringBuilder sb = new StringBuilder();
-      sb.append("#").append(id).append(", table=").append(tableName).append(", 
")
-        .append("attempt=").append(numAttempt)
-        .append("/").append(numTries).append(" ");
-
-      if (failureCount > 0 || error != null){
-        sb.append("failed=").append(failureCount).append("ops").append(", last 
exception: ").
-            append(error == null ? "null" : error);
-      } else {
-        sb.append("succeeded");
-      }
-
-      sb.append(" on ").append(sn).append(", tracking started 
").append(startTime);
-
-      if (willRetry) {
-        sb.append(", retrying after=").append(backOffTime).append("ms").
-            append(", replay=").append(replaySize).append("ops");
-      } else if (failureCount > 0) {
-        if (stopped > 0) {
-          sb.append("; not retrying ").append(stopped).append(" due to success 
from other replica");
-        }
-        if (failed > 0) {
-          sb.append("; not retrying ").append(failed).append(" - final 
failure");
-        }
-
-      }
-
-      return sb.toString();
-    }
-
-    /**
-     * Sets the non-error result from a particular action.
-     * @param action Action (request) that the server responded to.
-     * @param result The result.
-     */
-    private void setResult(Action<Row> action, Object result) {
-      if (result == null) {
-        throw new RuntimeException("Result cannot be null");
-      }
-      ReplicaResultState state = null;
-      boolean isStale = 
!RegionReplicaUtil.isDefaultReplica(action.getReplicaId());
-      int index = action.getOriginalIndex();
-      if (results == null) {
-        decActionCounter(index);
-        return; // Simple case, no replica requests.
-      }
-      state = trySetResultSimple(index, action.getAction(), false, result, 
null, isStale);
-      if (state == null) {
-        return; // Simple case, no replica requests.
-      }
-      // At this point we know that state is set to replica tracking class.
-      // It could be that someone else is also looking at it; however, we know 
there can
-      // only be one state object, and only one thread can set callCount to 0. 
Other threads
-      // will either see state with callCount 0 after locking it; or will not 
see state at all
-      // we will replace it with the result.
-      synchronized (state) {
-        if (state.callCount == 0) {
-          return; // someone already set the result
-        }
-        state.callCount = 0;
-      }
-      synchronized (replicaResultLock) {
-        if (results[index] != state) {
-          throw new AssertionError("We set the callCount but someone else 
replaced the result");
-        }
-        results[index] = result;
-      }
-
-      decActionCounter(index);
-    }
-
-    /**
-     * Sets the error from a particular action.
-     * @param index Original action index.
-     * @param row Original request.
-     * @param throwable The resulting error.
-     * @param server The source server.
-     */
-    private void setError(int index, Row row, Throwable throwable, ServerName 
server) {
-      ReplicaResultState state = null;
-      if (results == null) {
-        // Note that we currently cannot have replica requests with null 
results. So it shouldn't
-        // happen that multiple replica calls will call dAC for same actions 
with results == null.
-        // Only one call per action should be present in this case.
-        errors.add(throwable, row, server);
-        decActionCounter(index);
-        return; // Simple case, no replica requests.
-      }
-      state = trySetResultSimple(index, row, true, throwable, server, false);
-      if (state == null) {
-        return; // Simple case, no replica requests.
-      }
-      BatchErrors target = null; // Error will be added to final errors, or 
temp replica errors.
-      boolean isActionDone = false;
-      synchronized (state) {
-        switch (state.callCount) {
-          case 0: return; // someone already set the result
-          case 1: { // All calls failed, we are the last error.
-            target = errors;
-            isActionDone = true;
-            break;
-          }
-          default: {
-            assert state.callCount > 1;
-            if (state.replicaErrors == null) {
-              state.replicaErrors = new BatchErrors();
-            }
-            target = state.replicaErrors;
-            break;
-          }
-        }
-        --state.callCount;
-      }
-      target.add(throwable, row, server);
-      if (isActionDone) {
-        if (state.replicaErrors != null) { // last call, no need to lock
-          errors.merge(state.replicaErrors);
-        }
-        // See setResult for explanations.
-        synchronized (replicaResultLock) {
-          if (results[index] != state) {
-            throw new AssertionError("We set the callCount but someone else 
replaced the result");
-          }
-          results[index] = throwable;
-        }
-        decActionCounter(index);
-      }
-    }
-
-    /**
-     * 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.
-     * @param index Original action index.
-     * @param row Original request.
-     */
-    private boolean isActionComplete(int index, Row row) {
-      if (!isReplicaGet(row)) return false;
-      Object resObj = results[index];
-      return (resObj != null) && (!(resObj instanceof ReplicaResultState)
-          || ((ReplicaResultState)resObj).callCount == 0);
-    }
-
-    /**
-     * Tries to set the result or error for a particular action as if there 
were no replica calls.
-     * @return null if successful; replica state if there were in fact replica 
calls.
-     */
-    private ReplicaResultState trySetResultSimple(int index, Row row, boolean 
isError,
-        Object result, ServerName server, boolean isFromReplica) {
-      Object resObj = null;
-      if (!isReplicaGet(row)) {
-        if (isFromReplica) {
-          throw new AssertionError("Unexpected stale result for " + row);
-        }
-        results[index] = result;
-      } else {
-        synchronized (replicaResultLock) {
-          resObj = results[index];
-          if (resObj == null) {
-            if (isFromReplica) {
-              throw new AssertionError("Unexpected stale result for " + row);
-            }
-            results[index] = result;
-          }
-        }
-      }
-
-      ReplicaResultState rrs =
-          (resObj instanceof ReplicaResultState) ? (ReplicaResultState)resObj 
: null;
-      if (rrs == null && isError) {
-        // The resObj is not replica state (null or already set).
-        errors.add((Throwable)result, row, server);
-      }
-
-      if (resObj == null) {
-        // resObj is null - no replica calls were made.
-        decActionCounter(index);
-        return null;
-      }
-      return rrs;
-    }
-
-    private void decActionCounter(int index) {
-      long actionsRemaining = actionsInProgress.decrementAndGet();
-      if (actionsRemaining < 0) {
-        String error = buildDetailedErrorMsg("Incorrect actions in progress", 
index);
-        throw new AssertionError(error);
-      } else if (actionsRemaining == 0) {
-        synchronized (actionsInProgress) {
-          actionsInProgress.notifyAll();
-        }
-      }
-    }
-
-    private String buildDetailedErrorMsg(String string, int index) {
-      StringBuilder error = new StringBuilder(128);
-      error.append(string).append("; called for ").append(index).append(", 
actionsInProgress ")
-          .append(actionsInProgress.get()).append("; replica gets: ");
-      if (replicaGetIndices != null) {
-        for (int i = 0; i < replicaGetIndices.length; ++i) {
-          error.append(replicaGetIndices[i]).append(", ");
-        }
-      } else {
-        error.append(hasAnyReplicaGets ? "all" : "none");
-      }
-      error.append("; results ");
-      if (results != null) {
-        for (int i = 0; i < results.length; ++i) {
-          Object o = results[i];
-          error.append(((o == null) ? "null" : o.toString())).append(", ");
-        }
-      }
-      return error.toString();
-    }
-
-    @Override
-    public void waitUntilDone() throws InterruptedIOException {
-      try {
-        waitUntilDone(Long.MAX_VALUE);
-      } catch (InterruptedException iex) {
-        throw new InterruptedIOException(iex.getMessage());
-      } finally {
-        if (callsInProgress != null) {
-          for (CancellableRegionServerCallable clb : callsInProgress) {
-            clb.cancel();
-          }
-        }
-      }
-    }
-
-    private boolean waitUntilDone(long cutoff) throws InterruptedException {
-      boolean hasWait = cutoff != Long.MAX_VALUE;
-      long lastLog = EnvironmentEdgeManager.currentTime();
-      long currentInProgress;
-      while (0 != (currentInProgress = actionsInProgress.get())) {
-        long now = EnvironmentEdgeManager.currentTime();
-        if (hasWait && (now * 1000L) > cutoff) {
-          return false;
-        }
-        if (!hasWait) { // Only log if wait is infinite.
-          if (now > lastLog + 10000) {
-            lastLog = now;
-            LOG.info("#" + id + ", waiting for " + currentInProgress
-                + "  actions to finish on table: " + tableName);
-            if (currentInProgress <= thresholdToLogUndoneTaskDetails) {
-              logDetailsOfUndoneTasks(currentInProgress);
-            }
-          }
-        }
-        synchronized (actionsInProgress) {
-          if (actionsInProgress.get() == 0) break;
-          if (!hasWait) {
-            actionsInProgress.wait(10);
-          } else {
-            long waitMicroSecond = Math.min(100000L, (cutoff - now * 1000L));
-            TimeUnit.MICROSECONDS.timedWait(actionsInProgress, 
waitMicroSecond);
-          }
-        }
-      }
-      return true;
-    }
-
-    @Override
-    public boolean hasError() {
-      return errors.hasErrors();
-    }
-
-    @Override
-    public List<? extends Row> getFailedOperations() {
-      return errors.actions;
-    }
-
-    @Override
-    public RetriesExhaustedWithDetailsException getErrors() {
-      return errors.makeException(logBatchErrorDetails);
-    }
-
-    @Override
-    public Object[] getResults() throws InterruptedIOException {
-      waitUntilDone();
-      return results;
-    }
-  }
-
-  @VisibleForTesting
-  protected void updateStats(ServerName server, Map<byte[], 
MultiResponse.RegionResult> results) {
-    boolean metrics = AsyncProcess.this.connection.getConnectionMetrics() != 
null;
-    boolean stats = AsyncProcess.this.connection.getStatisticsTracker() != 
null;
-    if (!stats && !metrics) {
-      return;
-    }
-    for (Map.Entry<byte[], MultiResponse.RegionResult> regionStats : 
results.entrySet()) {
-      byte[] regionName = regionStats.getKey();
-      ClientProtos.RegionLoadStats stat = regionStats.getValue().getStat();
-      RegionLoadStats regionLoadstats = 
ProtobufUtil.createRegionLoadStats(stat);
-      
ResultStatsUtil.updateStats(AsyncProcess.this.connection.getStatisticsTracker(),
 server,
-          regionName, regionLoadstats);
-      
ResultStatsUtil.updateStats(AsyncProcess.this.connection.getConnectionMetrics(),
-          server, regionName, regionLoadstats);
-    }
-  }
-
   protected <CResult> AsyncRequestFutureImpl<CResult> createAsyncRequestFuture(
       TableName tableName, List<Action<Row>> actions, long nonceGroup, 
ExecutorService pool,
       Batch.Callback<CResult> callback, Object[] results, boolean needResults,
       CancellableRegionServerCallable callable, int curTimeout) {
     return new AsyncRequestFutureImpl<CResult>(
         tableName, actions, nonceGroup, getPool(pool), needResults,
-        results, callback, callable, curTimeout);
-  }
-
-  /**
-   * Create a callable. Isolated to be easily overridden in the tests.
-   */
-  @VisibleForTesting
-  protected MultiServerCallable<Row> createCallable(final ServerName server,
-      TableName tableName, final MultiAction<Row> multi) {
-    return new MultiServerCallable<Row>(connection, tableName, server, 
this.rpcFactory, multi);
-  }
-
-  /**
-   * Create a caller. Isolated to be easily overridden in the tests.
-   */
-  @VisibleForTesting
-  protected RpcRetryingCaller<AbstractResponse> createCaller(
-      CancellableRegionServerCallable callable) {
-    return rpcCallerFactory.<AbstractResponse> newCaller();
-  }
-
-  @VisibleForTesting
-  /** Waits until all outstanding tasks are done. Used in tests. */
-  void waitUntilDone() throws InterruptedIOException {
-    waitForMaximumCurrentTasks(0, null);
+        results, callback, callable, curTimeout, this);
   }
 
   /** Wait until the async does not have more than max tasks in progress. */
-  private void waitForMaximumCurrentTasks(int max, String tableName)
+  protected void waitForMaximumCurrentTasks(int max, String tableName)
       throws InterruptedIOException {
     waitForMaximumCurrentTasks(max, tasksInProgress, id, tableName);
   }
@@ -1839,7 +564,7 @@ class AsyncProcess {
     }
   }
 
-  private void logDetailsOfUndoneTasks(long taskInProgress) {
+  void logDetailsOfUndoneTasks(long taskInProgress) {
     ArrayList<ServerName> servers = new ArrayList<ServerName>();
     for (Map.Entry<ServerName, AtomicInteger> entry : 
taskCounterPerServer.entrySet()) {
       if (entry.getValue().get() > 0) {
@@ -1861,7 +586,7 @@ class AsyncProcess {
   /**
    * Only used w/useGlobalErrors ctor argument, for HTable backward compat.
    * @return Whether there were any errors in any request since the last time
-   *          {@link #waitForAllPreviousOpsAndReset(List)} was called, or AP 
was created.
+   *          {@link #waitForAllPreviousOpsAndReset(List, String)} was called, 
or AP was created.
    */
   public boolean hasError() {
     return globalErrors.hasErrors();
@@ -1872,9 +597,9 @@ class AsyncProcess {
    * Waits for all previous operations to finish, and returns errors and 
(optionally)
    * failed operations themselves.
    * @param failedRows an optional list into which the rows that failed since 
the last time
-   *        {@link #waitForAllPreviousOpsAndReset(List)} was called, or AP was 
created, are saved.
+   *        {@link #waitForAllPreviousOpsAndReset(List, String)} was called, 
or AP was created, are saved.
    * @param tableName name of the table
-   * @return all the errors since the last time {@link 
#waitForAllPreviousOpsAndReset(List)}
+   * @return all the errors since the last time {@link 
#waitForAllPreviousOpsAndReset(List, String)}
    *          was called, or AP was created.
    */
   public RetriesExhaustedWithDetailsException waitForAllPreviousOpsAndReset(
@@ -1934,6 +659,16 @@ class AsyncProcess {
   }
 
   /**
+   * Create a caller. Isolated to be easily overridden in the tests.
+   */
+  @VisibleForTesting
+  protected RpcRetryingCaller<AbstractResponse> createCaller(
+      CancellableRegionServerCallable callable) {
+    return rpcCallerFactory.<AbstractResponse> newCaller();
+  }
+
+
+  /**
    * 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.
@@ -1945,23 +680,11 @@ class AsyncProcess {
         this.serverTrackerTimeout, this.numTries);
   }
 
-  private static boolean isReplicaGet(Row row) {
+  static boolean isReplicaGet(Row row) {
     return (row instanceof Get) && (((Get)row).getConsistency() == 
Consistency.TIMELINE);
   }
 
   /**
-   * For {@link AsyncRequestFutureImpl#manageError(int, Row, Retry, Throwable, 
ServerName)}. Only
-   * used to make logging more clear, we don't actually care why we don't 
retry.
-   */
-  private enum Retry {
-    YES,
-    NO_LOCATION_PROBLEM,
-    NO_NOT_RETRIABLE,
-    NO_RETRIES_EXHAUSTED,
-    NO_OTHER_SUCCEEDED
-  }
-
-  /**
    * Collect all advices from checkers and make the final decision.
    */
   @VisibleForTesting

http://git-wip-us.apache.org/repos/asf/hbase/blob/2cf8907d/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRequestFuture.java
----------------------------------------------------------------------
diff --git 
a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRequestFuture.java
 
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRequestFuture.java
new file mode 100644
index 0000000..539e234
--- /dev/null
+++ 
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRequestFuture.java
@@ -0,0 +1,40 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+import java.io.InterruptedIOException;
+import java.util.List;
+
+/**
+ * The context used to wait for results from one submit call.
+ * 1) If AsyncProcess is set to track errors globally, and not per call (for 
HTable puts),
+ *    then errors and failed operations in this object will reflect global 
errors.
+ * 2) If submit call is made with needResults false, results will not be saved.
+ *  */
+@InterfaceAudience.Private
+public interface AsyncRequestFuture {
+  public boolean hasError();
+  public RetriesExhaustedWithDetailsException getErrors();
+  public List<? extends Row> getFailedOperations();
+  public Object[] getResults() throws InterruptedIOException;
+  /** Wait until all tasks are executed, successfully or not. */
+  public void waitUntilDone() throws InterruptedIOException;
+}
\ No newline at end of file

Reply via email to