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}