rh客户端 发生 RemotingException 时，重试 (#1135)

* fix: make rheakv client retry on RemotingException

---------

Co-authored-by: yuanyuan.liu <yuanyuan.liu@signalplus.com>

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/client/DefaultRheaKVRpcService.java

@@ -24,6 +24,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alipay.remoting.exception.RemotingException;
 import com.alipay.sofa.jraft.Status;
 import com.alipay.sofa.jraft.rhea.client.failover.FailoverClosure;
 import com.alipay.sofa.jraft.rhea.client.pd.AbstractPlacementDriverClient;
@@ -139,7 +140,12 @@ public void complete(final Object result, final Throwable err) {
  closure.run(new Status(-1, "RPC failed with address: %s, response: %s", endpoint, response));
  }
  } else {
- closure.failure(err);
+ if (err instanceof RemotingException) {
+ closure.setError(Errors.RPC_CONNECTION_ERROR);
+ closure.run(new Status(-1, "RPC failed occur exception %s", err.getMessage()));
+ } else {
+ closure.failure(err);
+ }
  }
  }
 

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/client/pd/DefaultPlacementDriverRpcService.java

@@ -24,6 +24,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alipay.remoting.exception.RemotingException;
 import com.alipay.sofa.jraft.Status;
 import com.alipay.sofa.jraft.rhea.client.failover.FailoverClosure;
 import com.alipay.sofa.jraft.rhea.cmd.pd.BaseRequest;
@@ -110,7 +111,12 @@ public void complete(final Object result, final Throwable err) {
  closure.run(new Status(-1, "RPC failed with address: %s, response: %s", endpoint, response));
  }
  } else {
- closure.failure(err);
+ if (err instanceof RemotingException) {
+ closure.setError(Errors.RPC_CONNECTION_ERROR);
+ closure.run(new Status(-1, "RPC failed occur exception %s", err.getMessage()));
+ } else {
+ closure.failure(err);
+ }
  }
  }
 

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/errors/ConnectionFailureException.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alipay.sofa.jraft.rhea.errors;
+
+/**
+ * Rpc connection failure exception.
+ */
+public class ConnectionFailureException extends ApiException {
+ private static final long serialVersionUID = -5958618149334588246L;
+
+ public ConnectionFailureException() {
+ }
+
+ public ConnectionFailureException(String message) {
+ super(message);
+ }
+
+ public ConnectionFailureException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public ConnectionFailureException(Throwable cause) {
+ super(cause);
+ }
+}

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/errors/Errors.java

@@ -36,7 +36,8 @@
  * Do not add exceptions that occur only on the client or only on the server here.
  */
 public enum Errors {
- UNKNOWN_SERVER_ERROR(-1, "The server experienced an unexpected error when processing the request",
+
+ UNKNOWN_SERVER_ERROR(-1, "The server experienced an unexpected error when processing the request.",
  UnknownServerException::new),
 
  NONE(0, null, message -> null),
@@ -64,13 +65,13 @@ public enum Errors {
 
  INVALID_REGION_EPOCH(9, "Invalid region epoch (membership or version changed).", InvalidRegionEpochException::new),
 
- INVALID_STORE_STATS(10, "Placement driver: invalid store stats", InvalidStoreStatsException::new),
+ INVALID_STORE_STATS(10, "Placement driver: invalid store stats.", InvalidStoreStatsException::new),
 
- INVALID_REGION_STATS(11, "Placement driver: invalid region stats", InvalidStoreStatsException::new),
+ INVALID_REGION_STATS(11, "Placement driver: invalid region stats.", InvalidStoreStatsException::new),
 
- STORE_HEARTBEAT_OUT_OF_DATE(12, "The store heartbeat info is out of date", StoreHeartbeatOutOfDateException::new),
+ STORE_HEARTBEAT_OUT_OF_DATE(12, "The store heartbeat info is out of date.", StoreHeartbeatOutOfDateException::new),
 
- REGION_HEARTBEAT_OUT_OF_DATE(13, "The region heartbeat info is out of date", RegionHeartbeatOutOfDateException::new),
+ REGION_HEARTBEAT_OUT_OF_DATE(13, "The region heartbeat info is out of date.", RegionHeartbeatOutOfDateException::new),
 
  CALL_SELF_ENDPOINT_ERROR(14, "The usual reason is that the rpc call selected the self endpoint.",
  CallSelfEndpointException::new),
@@ -84,7 +85,10 @@ public enum Errors {
  + "The new region cannot be created.", RangeSplitFailException::new),
 
  TOO_SMALL_TO_SPLIT(18, "The region size is too small to split. See the server logs for more details.",
- RangeSplitFailException::new);
+ RangeSplitFailException::new),
+
+ RPC_CONNECTION_ERROR(19, "RPC connection disconnected.",
+ ConnectionFailureException::new);
 
  private interface ApiExceptionBuilder {
  ApiException build(final String message);

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/errors/ErrorsHelper.java

@@ -26,6 +26,7 @@ public static boolean isInvalidPeer(final Errors error) {
  return error == Errors.CALL_SELF_ENDPOINT_ERROR //
  || error == Errors.NOT_LEADER //
  || error == Errors.NO_REGION_FOUND //
+ || error == Errors.RPC_CONNECTION_ERROR //
  || error == Errors.LEADER_NOT_AVAILABLE;
  }