[JAV-193] loadbalance retry supports to custom Retry Policy

handlers/handler-loadbalance/src/main/java/io/servicecomb/loadbalance/LoadBalanceRetryHandler.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2017 Huawei Technologies Co., Ltd
+ *
+ * Licensed 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 io.servicecomb.loadbalance;
+
+import java.net.ConnectException;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import com.netflix.client.RetryHandler;
+import com.netflix.client.Utils;
+
+public class LoadBalanceRetryHandler implements RetryHandler {
+
+  private final static LoadBalanceRetryHandler INSTANCE = new LoadBalanceRetryHandler();
+
+  private int retrySameServer;
+  private int retryNextServer;
+  private boolean retryEnabled;
+
+  public static LoadBalanceRetryHandler getInstance() {
+    return INSTANCE;
+  }
+
+  private List<Class<? extends Throwable>> retryOnSameExceptions = Lists
+      .newArrayList(new Class[]{SocketTimeoutException.class, ConnectException.class});
+
+  private List<Class<? extends Throwable>> circuitRelated = Lists
+      .newArrayList(new Class[]{SocketException.class, SocketTimeoutException.class});
+
+  private LoadBalanceRetryHandler() {
+    this.retrySameServer = 0;
+    this.retryNextServer = 0;
+    this.retryEnabled = true;
+  }
+
+  public void setRetryRuntimeParams(int retrySameServer, int retryNextServer, boolean retryEnabled) {
+    this.retrySameServer = retrySameServer;
+    this.retryNextServer = retryNextServer;
+    this.retryEnabled = retryEnabled;
+  }
+
+  public void addRetryOnSameExceptions(List<Class<? extends Throwable>> retryOnSameExceptions) {
+    if (retryOnSameExceptions != null && retryOnSameExceptions.size() != 0) {
+      this.retryOnSameExceptions.addAll(retryOnSameExceptions);
+    }
+  }
+
+  public List<Class<? extends Throwable>> getRetryOnSameExceptions() {
+    return retryOnSameExceptions;
+  }
+
+  public boolean removeRetryOnSameException(Class<? extends Throwable> exception) {
+    return retryOnSameExceptions.remove(exception);
+  }
+
+  @Override
+  public boolean isRetriableException(Throwable e, boolean sameServer) {
+    return retryEnabled && (!sameServer || Utils.isPresentAsCause(e, retryOnSameExceptions));
+  }
+
+  @Override
+  public boolean isCircuitTrippingException(Throwable e) {
+    return Utils.isPresentAsCause(e, circuitRelated);
+  }
+
+  @Override
+  public int getMaxRetriesOnSameServer() {
+    return retrySameServer;
+  }
+
+  @Override
+  public int getMaxRetriesOnNextServer() {
+    return retryNextServer;
+  }
+
+
+}

handlers/handler-loadbalance/src/main/java/io/servicecomb/loadbalance/LoadbalanceHandler.java

@@ -30,7 +30,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.netflix.client.DefaultLoadBalancerRetryHandler;
 import com.netflix.loadbalancer.IRule;
 import com.netflix.loadbalancer.RoundRobinRule;
 import com.netflix.loadbalancer.Server;
@@ -52,9 +51,9 @@
 
 /**
  * 负载均衡处理链
- *
  */
 public class LoadbalanceHandler extends AbstractHandler {
+
   private static final Logger LOGGER = LoggerFactory.getLogger(LoadbalanceHandler.class);
 
   private static final ExecutorService RETRY_POOL = Executors.newCachedThreadPool(new ThreadFactory() {
@@ -102,12 +101,12 @@ public void handle(Invocation invocation, AsyncResponse asyncResp) throws Except
 
     // invocation是请求级别的，因此每次调用都需要设置一次
     lb.setInvocation(invocation);
-    final LoadBalancer choosenLB = lb;
+    final LoadBalancer chosenLB = lb;
 
     if (!Configuration.INSTANCE.isRetryEnabled(invocation.getMicroserviceName())) {
-      send(invocation, asyncResp, choosenLB);
+      send(invocation, asyncResp, chosenLB);
     } else {
-      sendWithRetry(invocation, asyncResp, choosenLB);
+      sendWithRetry(invocation, asyncResp, chosenLB);
     }
   }
 
@@ -144,32 +143,32 @@ protected void setTransactionControlFilter(LoadBalancer lb, String microserviceN
     }
   }
 
-  private void send(Invocation invocation, AsyncResponse asyncResp, final LoadBalancer choosenLB) throws Exception {
+  private void send(Invocation invocation, AsyncResponse asyncResp, final LoadBalancer chosenLB) throws Exception {
     long time = System.currentTimeMillis();
-    CseServer server = (CseServer) choosenLB.chooseServer(invocation);
+    CseServer server = (CseServer) chosenLB.chooseServer(invocation);
     if (null == server) {
       asyncResp.consumerFail(ExceptionUtils.lbAddressNotFound(invocation.getMicroserviceName(),
           invocation.getMicroserviceVersionRule(),
           invocation.getConfigTransportName()));
       return;
     }
     server.setLastVisitTime(time);
-    choosenLB.getLoadBalancerStats().incrementNumRequests(server);
+    chosenLB.getLoadBalancerStats().incrementNumRequests(server);
     invocation.setEndpoint(server.getEndpoint());
     invocation.next(resp -> {
       // this stats is for WeightedResponseTimeRule
-      choosenLB.getLoadBalancerStats().noteResponseTime(server, (System.currentTimeMillis() - time));
+      chosenLB.getLoadBalancerStats().noteResponseTime(server, (System.currentTimeMillis() - time));
       if (resp.isFailed()) {
-        choosenLB.getLoadBalancerStats().incrementSuccessiveConnectionFailureCount(server);
+        chosenLB.getLoadBalancerStats().incrementSuccessiveConnectionFailureCount(server);
       } else {
-        choosenLB.getLoadBalancerStats().incrementActiveRequestsCount(server);
+        chosenLB.getLoadBalancerStats().incrementActiveRequestsCount(server);
       }
       asyncResp.handle(resp);
     });
   }
 
   private void sendWithRetry(Invocation invocation, AsyncResponse asyncResp,
-      final LoadBalancer choosenLB) throws Exception {
+      final LoadBalancer chosenLB) throws Exception {
     long time = System.currentTimeMillis();
     // retry in loadbalance, 2.0 feature
     final int currentHandler = invocation.getHandlerIndex();
@@ -238,12 +237,15 @@ public void onExecutionFailed(ExecutionContext<Invocation> context, Throwable fi
     listeners.add(listener);
     ExecutionContext<Invocation> context = new ExecutionContext<>(invocation, null, null, null);
 
+    LoadBalanceRetryHandler.getInstance().setRetryRuntimeParams(
+        Configuration.INSTANCE.getRetryOnSame(invocation.getMicroserviceName()),
+        Configuration.INSTANCE.getRetryOnNext(invocation.getMicroserviceName()),
+        true);
+
     LoadBalancerCommand<Response> command = LoadBalancerCommand.<Response>builder()
-        .withLoadBalancer(choosenLB)
+        .withLoadBalancer(chosenLB)
         .withServerLocator(invocation)
-        .withRetryHandler(new DefaultLoadBalancerRetryHandler(
-            Configuration.INSTANCE.getRetryOnSame(invocation.getMicroserviceName()),
-            Configuration.INSTANCE.getRetryOnNext(invocation.getMicroserviceName()), true))
+        .withRetryHandler(LoadBalanceRetryHandler.getInstance())
         .withListeners(listeners)
         .withExecutionContext(context)
         .build();
@@ -253,19 +255,19 @@ public Observable<Response> call(Server s) {
         return Observable.create(f -> {
           try {
             ((CseServer) s).setLastVisitTime(time);
-            choosenLB.getLoadBalancerStats().incrementNumRequests(s);
+            chosenLB.getLoadBalancerStats().incrementNumRequests(s);
             invocation.setHandlerIndex(currentHandler); // for retry
             invocation.setEndpoint(((CseServer) s).getEndpoint());
             invocation.next(resp -> {
               if (resp.isFailed()) {
                 LOGGER.error("service call error, msg is {}, server is {} ",
                     ((Throwable) resp.getResult()).getMessage(),
                     s);
-                choosenLB.getLoadBalancerStats().incrementSuccessiveConnectionFailureCount(s);
+                chosenLB.getLoadBalancerStats().incrementSuccessiveConnectionFailureCount(s);
                 f.onError(resp.getResult());
               } else {
-                choosenLB.getLoadBalancerStats().incrementActiveRequestsCount(s);
-                choosenLB.getLoadBalancerStats().noteResponseTime(s,
+                chosenLB.getLoadBalancerStats().incrementActiveRequestsCount(s);
+                chosenLB.getLoadBalancerStats().noteResponseTime(s,
                     (System.currentTimeMillis() - time));
                 f.onNext(resp);
                 f.onCompleted();

handlers/handler-loadbalance/src/test/java/io/servicecomb/loadbalance/TestLoadBalanceRetryHandler.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2017 Huawei Technologies Co., Ltd
+ *
+ * Licensed 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 io.servicecomb.loadbalance;
+
+import java.net.ConnectException;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.core.IsCollectionContaining.hasItem;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+import io.servicecomb.swagger.invocation.exception.InvocationException;
+
+public class TestLoadBalanceRetryHandler {
+
+  private final class DummyException extends RuntimeException {
+
+    DummyException(String message, Throwable e) {
+      super(message, e);
+    }
+  }
+
+  @Test
+  public void testRetryOnSameOperations() {
+    List<Class<? extends Throwable>> appendExceptions = Lists
+        .newArrayList(new Class[]{InvocationException.class, IllegalAccessException.class});
+
+    LoadBalanceRetryHandler.getInstance().setRetryRuntimeParams(1, 2, true);
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance().getMaxRetriesOnSameServer(), is(1));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance().getMaxRetriesOnNextServer(), is(2));
+
+    LoadBalanceRetryHandler.getInstance().addRetryOnSameExceptions(appendExceptions);
+    List<Class<? extends Throwable>> retryOnSameExceptions = LoadBalanceRetryHandler.getInstance()
+        .getRetryOnSameExceptions();
+    Assert.assertThat(retryOnSameExceptions, hasItem(InvocationException.class));
+    Assert.assertThat(retryOnSameExceptions, hasItem(IllegalAccessException.class));
+
+    Assert.assertThat(retryOnSameExceptions.size(), is(4));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance().removeRetryOnSameException(InvocationException.class),
+        is(true));
+    Assert
+        .assertThat(LoadBalanceRetryHandler.getInstance().removeRetryOnSameException(SocketException.class), is(false));
+    Assert.assertThat(retryOnSameExceptions.size(), is(3));
+    Assert.assertThat(retryOnSameExceptions,
+        contains(SocketTimeoutException.class, ConnectException.class, IllegalAccessException.class));
+  }
+
+  @Test
+  public void testIsRetriableException() {
+    LoadBalanceRetryHandler.getInstance().setRetryRuntimeParams(1, 2, true);
+    Assert
+        .assertThat(LoadBalanceRetryHandler.getInstance().isRetriableException(new SocketException(), false), is(true));
+    Assert
+        .assertThat(LoadBalanceRetryHandler.getInstance().isRetriableException(new SocketException(), true), is(false));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance().isRetriableException(new SocketTimeoutException(), true),
+        is(true));
+    Assert
+        .assertThat(LoadBalanceRetryHandler.getInstance().isRetriableException(new ConnectException(), true), is(true));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance()
+        .isRetriableException(new DummyException("connect failed", new ConnectException()), true), is(true));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance()
+        .isRetriableException(new DummyException("socket timeout", new SocketTimeoutException()), true), is(true));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance()
+        .isRetriableException(new DummyException("illegal access", new IllegalAccessException()), true), is(false));
+
+    LoadBalanceRetryHandler.getInstance().setRetryRuntimeParams(1, 2, false);
+    Assert
+        .assertThat(LoadBalanceRetryHandler.getInstance().isRetriableException(new SocketException(), true), is(false));
+
+  }
+
+  @Test
+  public void testIsCircuitTrippingException() {
+    LoadBalanceRetryHandler.getInstance().setRetryRuntimeParams(1, 2, true);
+
+    Assert
+        .assertThat(LoadBalanceRetryHandler.getInstance().isCircuitTrippingException(new SocketException()), is(true));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance().isCircuitTrippingException(new SocketTimeoutException()),
+        is(true));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance().isCircuitTrippingException(new IllegalAccessException()),
+        is(false));
+
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance()
+        .isCircuitTrippingException(new DummyException("socket exception", new SocketException())), is(true));
+    Assert.assertThat(LoadBalanceRetryHandler.getInstance()
+        .isCircuitTrippingException(new DummyException("socket timeout", new SocketTimeoutException())), is(true));
+    Assert.assertThat(
+        LoadBalanceRetryHandler.getInstance()
+            .isCircuitTrippingException(new DummyException("illegal access", new IllegalAccessException())), is(false));
+  }
+
+}