[SCB-164] reconnect on crash

.travis.yml

@@ -23,6 +23,7 @@ language: java
 jdk:
 - oraclejdk8
 script:
+- set -e
 - mvn clean install -Pjacoco -Pdocker coveralls:report
 - mvn clean verify -f saga-demo -Pdemo -Pdocker -Ddocker.useColor=false -Ddocker.showLogs
 after_success:

alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java

@@ -67,7 +67,7 @@ public void onDisconnected(GrpcServiceConfig request, StreamObserver<GrpcAck> re
       callback.disconnect();
     }
 
-    responseObserver.onNext(GrpcAck.newBuilder().build());
+    responseObserver.onNext(ACK);
     responseObserver.onCompleted();
   }
 

alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java

@@ -149,7 +149,7 @@ public void closeStreamOnDisconnected() {
         omegaCallbacks.get(serviceConfig.getServiceName()).containsKey(serviceConfig.getInstanceId()),
         is(false));
 
-    assertThat(compensateResponseObserver.isCompleted(), is(true));
+    await().atMost(1, SECONDS).until(compensateResponseObserver::isCompleted);
   }
 
   @Test

omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java

@@ -20,6 +20,8 @@
 
 package org.apache.servicecomb.saga.omega.connector.grpc;
 
+import java.util.function.Function;
+
 import org.apache.servicecomb.saga.omega.context.ServiceConfig;
 import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
 import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
@@ -54,13 +56,14 @@ public GrpcClientMessageSender(
       MessageSerializer serializer,
       MessageDeserializer deserializer,
       ServiceConfig serviceConfig,
+      Function<MessageSender, Runnable> errorHandlerFactory,
       MessageHandler handler) {
     this.target = address;
     this.asyncEventService = TxEventServiceGrpc.newStub(channel);
     this.blockingEventService = TxEventServiceGrpc.newBlockingStub(channel);
     this.serializer = serializer;
 
-    this.compensateStreamObserver = new GrpcCompensateStreamObserver(handler, deserializer);
+    this.compensateStreamObserver = new GrpcCompensateStreamObserver(handler, errorHandlerFactory.apply(this), deserializer);
     this.serviceConfig = serviceConfig(serviceConfig.serviceName(), serviceConfig.instanceId());
   }
 

omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcCompensateStreamObserver.java

@@ -35,10 +35,12 @@ class GrpcCompensateStreamObserver implements StreamObserver<GrpcCompensateComma
   private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
   private final MessageHandler messageHandler;
+  private final Runnable errorHandler;
   private final MessageDeserializer deserializer;
 
-  GrpcCompensateStreamObserver(MessageHandler messageHandler, MessageDeserializer deserializer) {
+  GrpcCompensateStreamObserver(MessageHandler messageHandler, Runnable errorHandler, MessageDeserializer deserializer) {
     this.messageHandler = messageHandler;
+    this.errorHandler = errorHandler;
     this.deserializer = deserializer;
   }
 
@@ -58,6 +60,7 @@ public void onNext(GrpcCompensateCommand command) {
   @Override
   public void onError(Throwable t) {
     LOG.error("failed to process grpc compensate command.", t);
+    errorHandler.run();
   }
 
   @Override

omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java

@@ -18,6 +18,7 @@
 package org.apache.servicecomb.saga.omega.connector.grpc;
 
 import static java.util.Collections.emptyList;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
@@ -26,7 +27,11 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.function.Consumer;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Function;
 
 import org.apache.servicecomb.saga.omega.context.ServiceConfig;
 import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
@@ -45,11 +50,15 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
   private final Map<MessageSender, Long> senders = new HashMap<>();
   private final Collection<ManagedChannel> channels;
 
+  private final BlockingQueue<Runnable> pendingTasks = new LinkedBlockingQueue<>();
+  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
   public LoadBalancedClusterMessageSender(String[] addresses,
       MessageSerializer serializer,
       MessageDeserializer deserializer,
       ServiceConfig serviceConfig,
-      MessageHandler handler) {
+      MessageHandler handler,
+      int reconnectDelay) {
 
     if (addresses.length == 0) {
       throw new IllegalArgumentException("No reachable cluster address provided");
@@ -69,11 +78,15 @@ public LoadBalancedClusterMessageSender(String[] addresses,
               serializer,
               deserializer,
               serviceConfig,
+              errorHandlerFactory(),
               handler),
           0L);
     }
+
+    scheduleReconnectTask(reconnectDelay);
   }
 
+  // this is for test only
   LoadBalancedClusterMessageSender(MessageSender... messageSenders) {
     for (MessageSender sender : messageSenders) {
       senders.put(sender, 0L);
@@ -106,35 +119,53 @@ public void onDisconnected() {
 
   @Override
   public void close() {
+    scheduler.shutdown();
     channels.forEach(ManagedChannel::shutdownNow);
   }
 
   @Override
   public void send(TxEvent event) {
     boolean success = false;
     do {
-      try {
-        withFastestSender(messageSender -> {
-          // very large latency on exception
-          senders.put(messageSender, Long.MAX_VALUE);
+      MessageSender messageSender = fastestSender();
 
-          long startTime = System.nanoTime();
-          messageSender.send(event);
-          senders.put(messageSender, System.nanoTime() - startTime);
-        });
+      try {
+        long startTime = System.nanoTime();
+        messageSender.send(event);
+        senders.put(messageSender, System.nanoTime() - startTime);
 
         success = true;
       } catch (Exception e) {
         log.error("Retry sending event {} due to failure", event, e);
+
+        // very large latency on exception
+        senders.put(messageSender, Long.MAX_VALUE);
       }
     } while (!success && !Thread.currentThread().isInterrupted());
   }
 
-  private void withFastestSender(Consumer<MessageSender> consumer) {
-    senders.entrySet()
+  private MessageSender fastestSender() {
+    return senders.entrySet()
         .stream()
         .min(Comparator.comparingLong(Entry::getValue))
         .map(Entry::getKey)
-        .ifPresent(consumer);
+        .orElse(NO_OP_SENDER);
+  }
+
+  private void scheduleReconnectTask(int reconnectDelay) {
+    scheduler.scheduleWithFixedDelay(() -> {
+      try {
+        pendingTasks.take().run();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
+    }, 0, reconnectDelay, MILLISECONDS);
+  }
+
+  private Function<MessageSender, Runnable> errorHandlerFactory() {
+    return messageSender -> {
+      Runnable runnable = new PushBackReconnectRunnable(messageSender, senders, pendingTasks);
+      return () -> pendingTasks.offer(runnable);
+    };
   }
 }

omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java

@@ -0,0 +1,56 @@
+/*
+ * 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 org.apache.servicecomb.saga.omega.connector.grpc;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class PushBackReconnectRunnable implements Runnable {
+  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+  private final MessageSender messageSender;
+  private final Map<MessageSender, Long> senders;
+  private final BlockingQueue<Runnable> pendingTasks;
+
+  PushBackReconnectRunnable(
+      MessageSender messageSender,
+      Map<MessageSender, Long> senders,
+      BlockingQueue<Runnable> pendingTasks) {
+    this.messageSender = messageSender;
+    this.senders = senders;
+    this.pendingTasks = pendingTasks;
+  }
+
+  @Override
+  public void run() {
+    try {
+      log.info("Retry connecting to alpha at {}", messageSender.target());
+      messageSender.onDisconnected();
+      messageSender.onConnected();
+      senders.put(messageSender, 0L);
+      log.info("Retry connecting to alpha at {} is successful", messageSender.target());
+    } catch (Exception e) {
+      log.error("Failed to reconnect to alpha at {}", messageSender.target(), e);
+      pendingTasks.offer(this);
+    }
+  }
+}