001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.client;
019
020import static org.apache.hadoop.hbase.client.ConnectionUtils.resetController;
021import static org.apache.hadoop.hbase.client.ConnectionUtils.translateException;
022
023import java.util.ArrayList;
024import java.util.List;
025import java.util.Map;
026import java.util.Optional;
027import java.util.OptionalLong;
028import java.util.concurrent.CompletableFuture;
029import java.util.concurrent.TimeUnit;
030import java.util.function.Consumer;
031import java.util.function.Supplier;
032import org.apache.hadoop.hbase.DoNotRetryIOException;
033import org.apache.hadoop.hbase.HBaseServerException;
034import org.apache.hadoop.hbase.NotServingRegionException;
035import org.apache.hadoop.hbase.TableName;
036import org.apache.hadoop.hbase.TableNotEnabledException;
037import org.apache.hadoop.hbase.TableNotFoundException;
038import org.apache.hadoop.hbase.client.backoff.HBaseServerExceptionPauseManager;
039import org.apache.hadoop.hbase.exceptions.ScannerResetException;
040import org.apache.hadoop.hbase.ipc.HBaseRpcController;
041import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
042import org.apache.hadoop.hbase.util.FutureUtils;
043import org.apache.yetus.audience.InterfaceAudience;
044import org.slf4j.Logger;
045import org.slf4j.LoggerFactory;
046
047import org.apache.hbase.thirdparty.io.netty.util.Timer;
048
049@InterfaceAudience.Private
050public abstract class AsyncRpcRetryingCaller<T> {
051
052  private static final Logger LOG = LoggerFactory.getLogger(AsyncRpcRetryingCaller.class);
053
054  private final Timer retryTimer;
055
056  private final int priority;
057
058  private final long startNs;
059
060  private int tries = 1;
061
062  private final int maxAttempts;
063
064  private final int startLogErrorsCnt;
065
066  private final List<RetriesExhaustedException.ThrowableWithExtraContext> exceptions;
067
068  private final long rpcTimeoutNs;
069
070  protected final long operationTimeoutNs;
071
072  protected final AsyncConnectionImpl conn;
073
074  protected final CompletableFuture<T> future;
075
076  protected final HBaseRpcController controller;
077
078  private final HBaseServerExceptionPauseManager pauseManager;
079
080  public AsyncRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn, int priority,
081    long pauseNs, long pauseNsForServerOverloaded, int maxAttempts, long operationTimeoutNs,
082    long rpcTimeoutNs, int startLogErrorsCnt, Map<String, byte[]> requestAttributes) {
083    this.retryTimer = retryTimer;
084    this.conn = conn;
085    this.priority = priority;
086    this.maxAttempts = maxAttempts;
087    this.operationTimeoutNs = operationTimeoutNs;
088    this.rpcTimeoutNs = rpcTimeoutNs;
089    this.startLogErrorsCnt = startLogErrorsCnt;
090    this.future = new CompletableFuture<>();
091    this.controller = conn.rpcControllerFactory.newController();
092    this.controller.setPriority(priority);
093    this.controller.setRequestAttributes(requestAttributes);
094    this.exceptions = new ArrayList<>();
095    this.startNs = System.nanoTime();
096    this.pauseManager =
097      new HBaseServerExceptionPauseManager(pauseNs, pauseNsForServerOverloaded, operationTimeoutNs);
098  }
099
100  private long elapsedMs() {
101    return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
102  }
103
104  protected final long remainingTimeNs() {
105    return pauseManager.remainingTimeNs(startNs);
106  }
107
108  protected final void completeExceptionally() {
109    future.completeExceptionally(new RetriesExhaustedException(tries - 1, exceptions));
110  }
111
112  protected final void resetCallTimeout() {
113    long callTimeoutNs;
114    if (operationTimeoutNs > 0) {
115      callTimeoutNs = remainingTimeNs();
116      if (callTimeoutNs <= 0) {
117        completeExceptionally();
118        return;
119      }
120      callTimeoutNs = Math.min(callTimeoutNs, rpcTimeoutNs);
121    } else {
122      callTimeoutNs = rpcTimeoutNs;
123    }
124    resetController(controller, callTimeoutNs, priority, getTableName().orElse(null));
125  }
126
127  private void tryScheduleRetry(Throwable error) {
128    OptionalLong maybePauseNsToUse = pauseManager.getPauseNsFromException(error, tries, startNs);
129    if (!maybePauseNsToUse.isPresent()) {
130      completeExceptionally();
131      return;
132    }
133    long delayNs = maybePauseNsToUse.getAsLong();
134    tries++;
135    if (HBaseServerException.isServerOverloaded(error)) {
136      Optional<MetricsConnection> metrics = conn.getConnectionMetrics();
137      metrics.ifPresent(m -> m.incrementServerOverloadedBackoffTime(delayNs, TimeUnit.NANOSECONDS));
138    }
139    retryTimer.newTimeout(t -> doCall(), delayNs, TimeUnit.NANOSECONDS);
140  }
141
142  protected Optional<TableName> getTableName() {
143    return Optional.empty();
144  }
145
146  // Sub classes can override this method to change the error type, to control the retry logic.
147  // For example, during rolling upgrading, if we call this newly added method, we will get a
148  // UnsupportedOperationException(wrapped by a DNRIOE), and sometimes we may want to fallback to
149  // use the old method first, so the sub class could change the exception type to something not a
150  // DNRIOE, so we will schedule a retry, and the next time the sub class could use old method to
151  // make the rpc call.
152  protected Throwable preProcessError(Throwable error) {
153    return error;
154  }
155
156  protected final void onError(Throwable t, Supplier<String> errMsg,
157    Consumer<Throwable> updateCachedLocation) {
158    if (future.isDone()) {
159      // Give up if the future is already done, this is possible if user has already canceled the
160      // future. And for timeline consistent read, we will also cancel some requests if we have
161      // already get one of the responses.
162      LOG.debug("The future is already done, canceled={}, give up retrying", future.isCancelled());
163      return;
164    }
165    Throwable error = preProcessError(translateException(t));
166    // We use this retrying caller to open a scanner, as it is idempotent, but we may throw
167    // ScannerResetException, which is a DoNotRetryIOException when opening a scanner as now we will
168    // also fetch data when opening a scanner. The intention here is that if we hit a
169    // ScannerResetException when scanning then we should try to open a new scanner, instead of
170    // retrying on the old one, so it is declared as a DoNotRetryIOException. But here we are
171    // exactly trying to open a new scanner, so we should retry on ScannerResetException.
172    if (error instanceof DoNotRetryIOException && !(error instanceof ScannerResetException)) {
173      future.completeExceptionally(error);
174      return;
175    }
176    if (tries > startLogErrorsCnt) {
177      LOG.warn(errMsg.get() + ", tries = " + tries + ", maxAttempts = " + maxAttempts
178        + ", timeout = " + TimeUnit.NANOSECONDS.toMillis(operationTimeoutNs)
179        + " ms, time elapsed = " + elapsedMs() + " ms", error);
180    }
181    updateCachedLocation.accept(error);
182    RetriesExhaustedException.ThrowableWithExtraContext qt =
183      new RetriesExhaustedException.ThrowableWithExtraContext(error,
184        EnvironmentEdgeManager.currentTime(), "");
185    exceptions.add(qt);
186    if (tries >= maxAttempts) {
187      completeExceptionally();
188      return;
189    }
190    // check whether the table has been disabled, notice that the check will introduce a request to
191    // meta, so here we only check for disabled for some specific exception types.
192    if (error instanceof NotServingRegionException || error instanceof RegionOfflineException) {
193      Optional<TableName> tableName = getTableName();
194      if (tableName.isPresent()) {
195        FutureUtils.addListener(conn.getAdmin().isTableDisabled(tableName.get()), (disabled, e) -> {
196          if (e != null) {
197            if (e instanceof TableNotFoundException) {
198              future.completeExceptionally(e);
199            } else {
200              // failed to test whether the table is disabled, not a big deal, continue retrying
201              tryScheduleRetry(error);
202            }
203            return;
204          }
205          if (disabled) {
206            future.completeExceptionally(new TableNotEnabledException(tableName.get()));
207          } else {
208            tryScheduleRetry(error);
209          }
210        });
211      } else {
212        tryScheduleRetry(error);
213      }
214    } else {
215      tryScheduleRetry(error);
216    }
217  }
218
219  protected abstract void doCall();
220
221  CompletableFuture<T> call() {
222    doCall();
223    return future;
224  }
225}