You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ni...@apache.org on 2018/09/27 23:07:37 UTC
[incubator-servicecomb-saga] 02/04: SCB-909 Add fail fast mechanism
when all alpha is down.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 8a7dcbb0f42523252e2e5533b8dce7689280d3d9
Author: cherrylzhao <zh...@126.com>
AuthorDate: Tue Sep 25 15:48:27 2018 +0800
SCB-909 Add fail fast mechanism when all alpha is down.
---
.../connector/grpc/core/GrpcOnErrorHandler.java | 7 +++++-
.../grpc/saga/SagaLoadBalancedSenderTest.java | 14 +++++++----
.../grpc/tcc/TccLoadBalanceSenderTest.java | 27 +++++++++++++++++++++-
3 files changed, 42 insertions(+), 6 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
index 7ab02a1..65d231a 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
@@ -21,6 +21,7 @@ import com.google.common.base.Supplier;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
@@ -64,7 +65,11 @@ public class GrpcOnErrorHandler {
@Override
public MessageSender get() {
try {
- return reconnectedSenders.take();
+ MessageSender messageSender = reconnectedSenders.poll(5, TimeUnit.SECONDS);
+ if (null == messageSender) {
+ throw new OmegaException("Failed to get reconnected sender, all alpha server is down.");
+ }
+ return messageSender;
} catch (InterruptedException e) {
throw new OmegaException("Failed to get reconnected sender", e);
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalancedSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalancedSenderTest.java
index 193d7ec..58a49e4 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalancedSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalancedSenderTest.java
@@ -18,7 +18,7 @@
package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import static com.seanyinx.github.unit.scaffolding.AssertUtils.expectFailing;
-import static java.lang.Thread.State.WAITING;
+import static java.lang.Thread.State.TERMINATED;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.contains;
@@ -254,16 +254,20 @@ public class SagaLoadBalancedSenderTest extends SagaLoadBalancedSenderTestBase {
final Thread thread = new Thread(new Runnable() {
@Override
public void run() {
- messageSender.send(event);
+ try {
+ messageSender.send(event);
+ } catch (OmegaException ex) {
+ assertThat(ex.getMessage().endsWith("all alpha server is down."), is(true));
+ }
}
});
thread.start();
// we don't want to keep sending on cluster down
- await().atMost(2, SECONDS).until(new Callable<Boolean>() {
+ await().atMost(10, SECONDS).until(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
- return thread.isAlive() && thread.getState().equals(WAITING);
+ return thread.getState().equals(TERMINATED);
}
});
@@ -276,12 +280,14 @@ public class SagaLoadBalancedSenderTest extends SagaLoadBalancedSenderTestBase {
startServerOnPort(8080);
startServerOnPort(8090);
+ messageSender.send(event);
await().atMost(2, SECONDS).until(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
return connected.get(8080).size() == 2 || connected.get(8090).size() == 2;
}
});
+
await().atMost(2, SECONDS).until(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
index af1085a..f396e0e 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
@@ -35,6 +35,7 @@ import com.google.common.collect.Maps;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.netty.NettyServerBuilder;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Iterator;
import java.util.Map;
@@ -211,6 +212,31 @@ public class TccLoadBalanceSenderTest extends LoadBalanceSenderTestBase {
assertThat(eventsMap.get(8080).size(), is(3));
}
+ @Test
+ public void failFastWhenAllServerWasDown() throws IOException {
+ tccLoadBalanceSender.onConnected();
+ await().atMost(2, TimeUnit.SECONDS).until(new Callable<Boolean>() {
+ @Override
+ public Boolean call() {
+ return connected.get(8080).size() == 1 && connected.get(8090).size() == 1;
+ }
+ });
+ assertThat((connected.get(8080).size() == 1 && connected.get(8090).size() == 1), is(true));
+
+ for (Server each : servers.values()) {
+ each.shutdownNow();
+ }
+
+ try {
+ tccLoadBalanceSender.participate(participatedEvent);
+ } catch (OmegaException ex) {
+ assertThat(ex.getMessage().endsWith("all alpha server is down."), is(true));
+ }
+ for (Integer each : ports) {
+ startServerOnPort(each);
+ }
+ }
+
@Test(expected = OmegaException.class)
public void participateFailedThenAbort() {
TccMessageSender failedSender = mock(GrpcTccClientMessageSender.class);
@@ -285,5 +311,4 @@ public class TccLoadBalanceSenderTest extends LoadBalanceSenderTestBase {
Assert.assertThat(connected.get(8080), contains("Connected " + serviceName, "Disconnected " + serviceName));
Assert.assertThat(connected.get(8090), contains("Connected " + serviceName, "Disconnected " + serviceName));
}
-
}