001/**
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements.  See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership.  The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License.  You may obtain a copy of the License at
010 *
011 *     http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS,
015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
016 * See the License for the specific language governing permissions and
017 * limitations under the License.
018 */
019
020package org.apache.hadoop.hbase.client;
021
022import static org.apache.hadoop.hbase.client.ConnectionUtils.SLEEP_DELTA_NS;
023import static org.apache.hadoop.hbase.client.ConnectionUtils.getPauseTime;
024import static org.apache.hadoop.hbase.client.ConnectionUtils.resetController;
025import static org.apache.hadoop.hbase.client.ConnectionUtils.translateException;
026import org.apache.hbase.thirdparty.io.netty.util.HashedWheelTimer;
027
028import java.util.ArrayList;
029import java.util.List;
030import java.util.concurrent.CompletableFuture;
031import java.util.concurrent.TimeUnit;
032import java.util.function.Consumer;
033import java.util.function.Supplier;
034
035import org.apache.hadoop.hbase.DoNotRetryIOException;
036import org.apache.yetus.audience.InterfaceAudience;
037import org.slf4j.Logger;
038import org.slf4j.LoggerFactory;
039import org.apache.hadoop.hbase.ipc.HBaseRpcController;
040import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
041
042@InterfaceAudience.Private
043public abstract class AsyncRpcRetryingCaller<T> {
044
045  private static final Logger LOG = LoggerFactory.getLogger(AsyncRpcRetryingCaller.class);
046
047  private final HashedWheelTimer retryTimer;
048
049  private final long startNs;
050
051  private final long pauseNs;
052
053  private int tries = 1;
054
055  private final int maxAttempts;
056
057  private final int startLogErrorsCnt;
058
059  private final List<RetriesExhaustedException.ThrowableWithExtraContext> exceptions;
060
061  private final long rpcTimeoutNs;
062
063  protected final long operationTimeoutNs;
064
065  protected final AsyncConnectionImpl conn;
066
067  protected final CompletableFuture<T> future;
068
069  protected final HBaseRpcController controller;
070
071  public AsyncRpcRetryingCaller(HashedWheelTimer retryTimer, AsyncConnectionImpl conn,
072      long pauseNs, int maxAttempts, long operationTimeoutNs,
073      long rpcTimeoutNs, int startLogErrorsCnt) {
074    this.retryTimer = retryTimer;
075    this.conn = conn;
076    this.pauseNs = pauseNs;
077    this.maxAttempts = maxAttempts;
078    this.operationTimeoutNs = operationTimeoutNs;
079    this.rpcTimeoutNs = rpcTimeoutNs;
080    this.startLogErrorsCnt = startLogErrorsCnt;
081    this.future = new CompletableFuture<>();
082    this.controller = conn.rpcControllerFactory.newController();
083    this.exceptions = new ArrayList<>();
084    this.startNs = System.nanoTime();
085  }
086
087  private long elapsedMs() {
088    return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
089  }
090
091  protected long remainingTimeNs() {
092    return operationTimeoutNs - (System.nanoTime() - startNs);
093  }
094
095  protected void completeExceptionally() {
096    future.completeExceptionally(new RetriesExhaustedException(tries - 1, exceptions));
097  }
098
099  protected void resetCallTimeout() {
100    long callTimeoutNs;
101    if (operationTimeoutNs > 0) {
102      callTimeoutNs = remainingTimeNs();
103      if (callTimeoutNs <= 0) {
104        completeExceptionally();
105        return;
106      }
107      callTimeoutNs = Math.min(callTimeoutNs, rpcTimeoutNs);
108    } else {
109      callTimeoutNs = rpcTimeoutNs;
110    }
111    resetController(controller, callTimeoutNs);
112  }
113
114  protected void onError(Throwable error, Supplier<String> errMsg,
115      Consumer<Throwable> updateCachedLocation) {
116    error = translateException(error);
117    if (error instanceof DoNotRetryIOException) {
118      future.completeExceptionally(error);
119      return;
120    }
121    if (tries > startLogErrorsCnt) {
122      LOG.warn(errMsg.get() + ", tries = " + tries + ", maxAttempts = " + maxAttempts
123          + ", timeout = " + TimeUnit.NANOSECONDS.toMillis(operationTimeoutNs)
124          + " ms, time elapsed = " + elapsedMs() + " ms", error);
125    }
126    RetriesExhaustedException.ThrowableWithExtraContext qt = new RetriesExhaustedException.ThrowableWithExtraContext(
127        error, EnvironmentEdgeManager.currentTime(), "");
128    exceptions.add(qt);
129    if (tries >= maxAttempts) {
130      completeExceptionally();
131      return;
132    }
133    long delayNs;
134    if (operationTimeoutNs > 0) {
135      long maxDelayNs = remainingTimeNs() - SLEEP_DELTA_NS;
136      if (maxDelayNs <= 0) {
137        completeExceptionally();
138        return;
139      }
140      delayNs = Math.min(maxDelayNs, getPauseTime(pauseNs, tries - 1));
141    } else {
142      delayNs = getPauseTime(pauseNs, tries - 1);
143    }
144    updateCachedLocation.accept(error);
145    tries++;
146    retryTimer.newTimeout(t -> doCall(), delayNs, TimeUnit.NANOSECONDS);
147  }
148
149  protected abstract void doCall();
150
151  CompletableFuture<T> call() {
152    doCall();
153    return future;
154  }
155}