hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncMasterRequestRpcRetryingCaller.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.client;
  19
  20 import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
  21
  22 import java.util.concurrent.CompletableFuture;
  23 import org.apache.hadoop.hbase.exceptions.ClientExceptionsUtil;
  24 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
  25 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
  26 import org.apache.yetus.audience.InterfaceAudience;
  27
  28 import org.apache.hbase.thirdparty.io.netty.util.Timer;
  29
  30 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MasterService;
  31
  32 /**
  33  * Retry caller for a request call to master.
  34  * @since 2.0.0
  35  */
  36 @InterfaceAudience.Private
  37 public class AsyncMasterRequestRpcRetryingCaller<T> extends AsyncRpcRetryingCaller<T> {
  38
  39   @FunctionalInterface
  40   public interface Callable<T> {
  41     CompletableFuture<T> call(HBaseRpcController controller, MasterService.Interface stub);
  42   }
  43
  44   private final Callable<T> callable;
  45
  46   public AsyncMasterRequestRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn,
  47       Callable<T> callable, int priority, long pauseNs, long pauseForCQTBENs, int maxRetries,
  48       long operationTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt) {
  49     super(retryTimer, conn, priority, pauseNs, pauseForCQTBENs, maxRetries, operationTimeoutNs,
  50       rpcTimeoutNs, startLogErrorsCnt);
  51     this.callable = callable;
  52   }
  53
  54   private void clearMasterStubCacheOnError(MasterService.Interface stub, Throwable error) {
  55     // ServerNotRunningYetException may because it is the backup master.
  56     if (ClientExceptionsUtil.isConnectionException(error) ||
  57       error instanceof ServerNotRunningYetException) {
  58       conn.clearMasterStubCache(stub);
  59     }
  60   }
  61
  62   @Override
  63   protected void doCall() {
  64     addListener(conn.getMasterStub(), (stub, error) -> {
  65       if (error != null) {
  66         onError(error, () -> "Get async master stub failed", err -> {
  67         });
  68         return;
  69       }
  70       resetCallTimeout();
  71       addListener(callable.call(controller, stub), (result, error2) -> {
  72         if (error2 != null) {
  73           onError(error2, () -> "Call to master failed",
  74             err -> clearMasterStubCacheOnError(stub, error2));
  75           return;
  76         }
  77         future.complete(result);
  78       });
  79     });
  80   }
  81 }