HBase client 客户端重试机制

/**
* Retries if invocation fails.
* @param callTimeout Timeout for this call
* @param callable The {@link RetryingCallable} to run.
* @return an object of type T
* @throws IOException if a remote or network exception occurs
* @throws RuntimeException other unspecified error
*/
public T callWithRetries(RetryingCallable<T> callable, int callTimeout)
throws IOException, RuntimeException {
List<RetriesExhaustedException.ThrowableWithExtraContext> exceptions =
  new ArrayList<RetriesExhaustedException.ThrowableWithExtraContext>();
this.globalStartTime = EnvironmentEdgeManager.currentTime();
context.clear();
for (int tries = 0;; tries++) {
  long expectedSleep;
  try {
    callable.prepare(tries != 0); // if called with false, check table status on ZK
    interceptor.intercept(context.prepare(callable, tries));
    return callable.call(getRemainingTime(callTimeout));
  } catch (PreemptiveFastFailException e) {
    throw e;
  } catch (Throwable t) {
    ExceptionUtil.rethrowIfInterrupt(t);
    if (tries > startLogErrorsCnt) {
      LOG.info("Call exception, tries=" + tries + ", retries=" + retries + ", started=" +
          (EnvironmentEdgeManager.currentTime() - this.globalStartTime) + " ms ago, "
          + "cancelled=" + cancelled.get() + ", msg="
          + callable.getExceptionMessageAdditionalDetail());
    }

    // translateException throws exception when should not retry: i.e. when request is bad.
    interceptor.handleFailure(context, t);
    t = translateException(t);
    callable.throwable(t, retries != 1);
    RetriesExhaustedException.ThrowableWithExtraContext qt =
        new RetriesExhaustedException.ThrowableWithExtraContext(t,
            EnvironmentEdgeManager.currentTime(), toString());
    exceptions.add(qt);
    if (tries >= retries - 1) {
      throw new RetriesExhaustedException(tries, exceptions);
    }
    // If the server is dead, we need to wait a little before retrying, to give
    //  a chance to the regions to be
    // tries hasn't been bumped up yet so we use "tries + 1" to get right pause time
    expectedSleep = callable.sleep(pause, tries + 1);

    // If, after the planned sleep, there won't be enough time left, we stop now.
    long duration = singleCallDuration(expectedSleep);
    if (duration > callTimeout) {
      String msg = "callTimeout=" + callTimeout + ", callDuration=" + duration +
          ": " + callable.getExceptionMessageAdditionalDetail();
      throw (SocketTimeoutException)(new SocketTimeoutException(msg).initCause(t));
    }
  } finally {
    interceptor.updateFailureInfo(context);
  }
  try {
    if (expectedSleep > 0) {
      synchronized (cancelled) {
        if (cancelled.get()) return null;
        cancelled.wait(expectedSleep);
      }
    }
    if (cancelled.get()) return null;
  } catch (InterruptedException e) {
    throw new InterruptedIOException("Interrupted after " + tries + " tries  on " + retries);
  }
}
}

public static int RETRY_BACKOFF[] = { 1, 2, 3, 5, 10, 20, 40, 100, 100, 100, 100, 200, 200 }; long normalPause = pause * HConstants.RETRY_BACKOFF[ntries];long jitter = (long)(normalPause * RANDOM.nextFloat() * 0.01f);