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/01/10 03:31:24 UTC
[incubator-servicecomb-saga] branch master updated (b93e795 ->
367b1cc)
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git.
from b93e795 SCB-173 move context init from request to preInterceptor
new e5efd41 SCB-168 load balanced alpha cluster client
new 01c4a1d SCB-168 load balance based on server latency
new 2c395da SCB-168 exception free connection & disconnection
new 5529689 SCB-168 made latency significant enough to pass test
new e2828ee SCB-168 resolved rebase conflict
new 1e7ff02 SCB-168 removed default logging from spring boot
new 367b1cc SCB-168 connected to alpha on start
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../saga/integration/pack/tests/PackIT.java | 4 +-
omega/omega-connector/omega-connector-grpc/pom.xml | 19 ++
.../connector/grpc/GrpcClientMessageSender.java | 11 +-
.../grpc/LoadBalancedClusterMessageSender.java | 140 ++++++++++
.../grpc/LoadBalancedClusterMessageSenderTest.java | 295 +++++++++++++++++++++
.../src/test/resources/log4j2-test.xml | 0
.../saga/omega/format/MessageFormat.java | 12 +-
.../saga/omega/format/KryoMessageFormatTest.java | 6 +-
.../saga/omega/format/MessageFormatTestBase.java | 9 +-
.../saga/omega/spring/OmegaSpringConfig.java | 69 ++---
omega/omega-spring-tx/pom.xml | 6 +
.../saga/omega/transaction/MessageSender.java | 7 +
.../saga/omega/transaction/MessageSerializer.java | 2 -
13 files changed, 505 insertions(+), 75 deletions(-)
create mode 100644 omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
create mode 100644 omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
copy {saga-spring => omega/omega-connector/omega-connector-grpc}/src/test/resources/log4j2-test.xml (100%)
--
To stop receiving notification emails like this one, please contact
['"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>'].
[incubator-servicecomb-saga] 03/07: SCB-168 exception free
connection & disconnection
Posted by ni...@apache.org.
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 2c395dab2f19125738d5132ac51ba27b45559ae8
Author: seanyinx <se...@huawei.com>
AuthorDate: Mon Jan 8 16:22:21 2018 +0800
SCB-168 exception free connection & disconnection
Signed-off-by: seanyinx <se...@huawei.com>
---
.../connector/grpc/GrpcClientMessageSender.java | 11 +++++++-
.../grpc/LoadBalancedClusterMessageSender.java | 21 ++++++++++++---
.../grpc/LoadBalancedClusterMessageSenderTest.java | 31 ++++++++++++++++++++++
.../saga/omega/transaction/MessageSender.java | 4 +++
4 files changed, 63 insertions(+), 4 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
index 59fbce1..4729db1 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
@@ -39,6 +39,7 @@ import io.grpc.ManagedChannel;
public class GrpcClientMessageSender implements MessageSender {
+ private final String target;
private final TxEventServiceStub asyncEventService;
private final MessageSerializer serializer;
@@ -47,11 +48,14 @@ public class GrpcClientMessageSender implements MessageSender {
private final GrpcCompensateStreamObserver compensateStreamObserver;
private final GrpcServiceConfig serviceConfig;
- public GrpcClientMessageSender(ManagedChannel channel,
+ public GrpcClientMessageSender(
+ String address,
+ ManagedChannel channel,
MessageSerializer serializer,
MessageDeserializer deserializer,
ServiceConfig serviceConfig,
MessageHandler handler) {
+ this.target = address;
this.asyncEventService = TxEventServiceGrpc.newStub(channel);
this.blockingEventService = TxEventServiceGrpc.newBlockingStub(channel);
this.serializer = serializer;
@@ -71,6 +75,11 @@ public class GrpcClientMessageSender implements MessageSender {
}
@Override
+ public String target() {
+ return target;
+ }
+
+ @Override
public void send(TxEvent event) {
blockingEventService.onTxEvent(convertEvent(event));
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
index e5c9d19..14ce340 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
@@ -63,7 +63,9 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
channels.add(channel);
senders.put(
- new GrpcClientMessageSender(channel,
+ new GrpcClientMessageSender(
+ address,
+ channel,
serializer,
deserializer,
serviceConfig,
@@ -81,12 +83,25 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
@Override
public void onConnected() {
- senders.keySet().forEach(MessageSender::onConnected);
+ senders.keySet().forEach(sender -> {
+ try {
+ sender.onConnected();
+ } catch (Exception e) {
+ log.error("Failed connecting to alpha at {}", sender.target(), e);
+ }
+ });
}
@Override
public void onDisconnected() {
- senders.keySet().forEach(MessageSender::onDisconnected);
+ senders.keySet().forEach(sender -> {
+ try {
+ sender.onDisconnected();
+ } catch (Exception e) {
+ log.error("Failed disconnecting from alpha at {}", sender.target(), e);
+ }
+ });
+
}
@Override
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
index 6f2d90b..dc75663 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -28,6 +28,7 @@ import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
import java.io.IOException;
import java.util.ArrayList;
@@ -184,6 +185,36 @@ public class LoadBalancedClusterMessageSenderTest {
}
@Test
+ public void swallowException_UntilAllSendersConnected() throws Exception {
+ MessageSender underlying1 = Mockito.mock(MessageSender.class);
+ doThrow(RuntimeException.class).when(underlying1).onConnected();
+
+ MessageSender underlying2 = Mockito.mock(MessageSender.class);
+
+ MessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2);
+
+ sender.onConnected();
+
+ verify(underlying1).onConnected();
+ verify(underlying2).onConnected();
+ }
+
+ @Test
+ public void swallowException_UntilAllSendersDisconnected() throws Exception {
+ MessageSender underlying1 = Mockito.mock(MessageSender.class);
+ doThrow(RuntimeException.class).when(underlying1).onDisconnected();
+
+ MessageSender underlying2 = Mockito.mock(MessageSender.class);
+
+ MessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2);
+
+ sender.onDisconnected();
+
+ verify(underlying1).onDisconnected();
+ verify(underlying2).onDisconnected();
+ }
+
+ @Test
public void considerFasterServerFirst() throws Exception {
// we don't know which server is selected at first
messageSender.send(event);
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
index 39b4d62..6a4ede6 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
@@ -27,5 +27,9 @@ public interface MessageSender {
default void close() {
}
+ default String target() {
+ return "UNKNOWN";
+ }
+
void send(TxEvent event);
}
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.
[incubator-servicecomb-saga] 01/07: SCB-168 load balanced alpha
cluster client
Posted by ni...@apache.org.
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 e5efd4164e5d8b35b1f336c1c6657d37fb3fe257
Author: seanyinx <se...@huawei.com>
AuthorDate: Mon Jan 8 14:22:51 2018 +0800
SCB-168 load balanced alpha cluster client
Signed-off-by: seanyinx <se...@huawei.com>
---
omega/omega-connector/omega-connector-grpc/pom.xml | 16 ++
.../grpc/LoadBalancedClusterMessageSender.java | 139 ++++++++++++++
.../grpc/LoadBalancedClusterMessageSenderTest.java | 211 +++++++++++++++++++++
.../src/test/resources/log4j2-test.xml | 30 +++
.../saga/omega/format/MessageFormatTestBase.java | 2 +-
.../saga/omega/transaction/MessageSerializer.java | 2 -
6 files changed, 397 insertions(+), 3 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/pom.xml b/omega/omega-connector/omega-connector-grpc/pom.xml
index 375fac7..258eb4b 100644
--- a/omega/omega-connector/omega-connector-grpc/pom.xml
+++ b/omega/omega-connector/omega-connector-grpc/pom.xml
@@ -53,6 +53,18 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ </dependency>
<dependency>
<groupId>junit</groupId>
@@ -70,5 +82,9 @@
<groupId>com.github.seanyinx</groupId>
<artifactId>unit-scaffolding</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ </dependency>
</dependencies>
</project>
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
new file mode 100644
index 0000000..276f887
--- /dev/null
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
@@ -0,0 +1,139 @@
+/*
+ * 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.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.servicecomb.saga.omega.context.ServiceConfig;
+import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
+import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.apache.servicecomb.saga.omega.transaction.MessageSerializer;
+import org.apache.servicecomb.saga.omega.transaction.TxEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.grpc.Attributes;
+import io.grpc.EquivalentAddressGroup;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.NameResolver;
+import io.grpc.NameResolver.Factory;
+import io.grpc.util.RoundRobinLoadBalancerFactory;
+
+public class LoadBalancedClusterMessageSender implements MessageSender {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private final MessageSender messageSender;
+
+ public LoadBalancedClusterMessageSender(String addresses,
+ MessageSerializer serializer,
+ MessageDeserializer deserializer,
+ ServiceConfig serviceConfig,
+ MessageHandler handler) {
+
+ this(new GrpcClientMessageSender(clusterDirectAddressChannel(addresses),
+ serializer,
+ deserializer,
+ serviceConfig,
+ handler));
+ }
+
+ LoadBalancedClusterMessageSender(MessageSender messageSender) {
+ this.messageSender = messageSender;
+ }
+
+ private static ManagedChannel clusterDirectAddressChannel(String addresses) {
+ return ManagedChannelBuilder.forTarget(addresses)
+ .nameResolverFactory(new ClusterNameResolverFactory(addresses))
+ .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
+ .usePlaintext(true)
+ .build();
+ }
+
+ @Override
+ public void onConnected() {
+ messageSender.onConnected();
+ }
+
+ @Override
+ public void onDisconnected() {
+ messageSender.onDisconnected();
+ }
+
+ @Override
+ public void send(TxEvent event) {
+ boolean success = false;
+ do {
+ try {
+ messageSender.send(event);
+ success = true;
+ } catch (Exception e) {
+ log.error("Retry sending event {} due to failure", event, e);
+ }
+ } while (!success && !Thread.currentThread().isInterrupted());
+
+ }
+
+ private static class ClusterNameResolverFactory extends Factory {
+ private final String addresses;
+
+ private ClusterNameResolverFactory(String addresses) {
+ this.addresses = addresses;
+ }
+
+ @Override
+ public NameResolver newNameResolver(URI targetUri, Attributes params) {
+ return new NameResolver() {
+ @Override
+ public String getServiceAuthority() {
+ return "localhost";
+ }
+
+ @Override
+ public void start(final Listener listener) {
+ List<SocketAddress> socketAddresses = Arrays.stream(addresses.split(","))
+ .map(address -> {
+ String[] split = address.split(":");
+ return new InetSocketAddress(split[0], Integer.parseInt(split[1]));
+ })
+ .collect(Collectors.toList());
+
+ listener.onAddresses(
+ Arrays.asList(new EquivalentAddressGroup(socketAddresses.get(0)),
+ new EquivalentAddressGroup(socketAddresses.get(1))),
+ Attributes.EMPTY);
+ }
+
+ @Override
+ public void shutdown() {
+ }
+ };
+ }
+
+ @Override
+ public String getDefaultScheme() {
+ return "directaddress";
+ }
+ }
+}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
new file mode 100644
index 0000000..4fee0e6
--- /dev/null
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -0,0 +1,211 @@
+/*
+ * 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 static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doThrow;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+import org.apache.servicecomb.saga.omega.context.ServiceConfig;
+import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
+import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.apache.servicecomb.saga.omega.transaction.MessageSerializer;
+import org.apache.servicecomb.saga.omega.transaction.TxEvent;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTxEvent;
+import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceImplBase;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.stub.StreamObserver;
+
+public class LoadBalancedClusterMessageSenderTest {
+
+ private static final int[] ports = {8080, 8090};
+ private static final List<Server> servers = new ArrayList<>();
+
+ private static final Queue<TxEvent> eventsOn8080 = new ConcurrentLinkedQueue<>();
+ private static final Queue<TxEvent> eventsOn8090 = new ConcurrentLinkedQueue<>();
+
+ private static final Map<Integer, Set<String>> connected = new HashMap<Integer, Set<String>>() {{
+ put(8080, new ConcurrentSkipListSet<>());
+ put(8090, new ConcurrentSkipListSet<>());
+ }};
+
+ private static final Map<Integer, Queue<TxEvent>> eventsMap = new HashMap<Integer, Queue<TxEvent>>() {{
+ put(8080, eventsOn8080);
+ put(8090, eventsOn8090);
+ }};
+
+ private final String addresses = "localhost:8080,localhost:8090";
+
+ private final MessageSerializer serializer = objects -> objects[0].toString().getBytes();
+
+ private final MessageDeserializer deserializer = message -> new Object[] {new String(message)};
+ private final MessageHandler handler = (globalTxId, localTxId, parentTxId, compensationMethod, payloads) -> {
+ };
+
+ private final String globalTxId = uniquify("globalTxId");
+ private final String localTxId = uniquify("localTxId");
+ private final String parentTxId = uniquify("parentTxId");
+ private final String compensationMethod = getClass().getCanonicalName();
+ private final TxEvent event = new TxEvent(globalTxId, localTxId, parentTxId, compensationMethod, "blah");
+
+ private final String serviceName = uniquify("serviceName");
+ private final MessageSender messageSender = new LoadBalancedClusterMessageSender(
+ addresses,
+ serializer,
+ deserializer,
+ new ServiceConfig(serviceName),
+ handler);
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ Arrays.stream(ports).forEach(port -> {
+ ServerBuilder<?> serverBuilder = ServerBuilder.forPort(port);
+ serverBuilder.addService(new MyTxEventService(connected.get(port), eventsMap.get(port)));
+ Server server = serverBuilder.build();
+
+ try {
+ server.start();
+ servers.add(server);
+ } catch (IOException e) {
+ fail(e.getMessage());
+ }
+ });
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ servers.forEach(Server::shutdown);
+ }
+
+ @Test
+ public void reconnectOnConnectionLoss() throws Exception {
+ messageSender.send(event);
+
+ killServerReceivedMessage();
+
+ messageSender.send(event);
+
+ assertThat(eventsOn8080.size(), is(1));
+ assertThat(eventsOn8080.peek().toString(), is(event.toString()));
+
+ assertThat(eventsOn8090.size(), is(1));
+ assertThat(eventsOn8090.peek().toString(), is(event.toString()));
+ }
+
+ @Test (timeout = 1000)
+ public void stopSendingOnInterruption() throws Exception {
+ MessageSender underlying = Mockito.mock(MessageSender.class);
+ doThrow(RuntimeException.class).when(underlying).send(event);
+
+ MessageSender messageSender = new LoadBalancedClusterMessageSender(underlying);
+
+ Thread thread = new Thread(() -> messageSender.send(event));
+ thread.start();
+
+ Thread.sleep(300);
+
+ thread.interrupt();
+ thread.join();
+ }
+
+ @Ignore
+ @Test
+ public void broadcastConnectionAndDisconnection() throws Exception {
+ messageSender.onConnected();
+ await().atMost(1, SECONDS).until(() -> !connected.get(8080).isEmpty() && !connected.get(8090).isEmpty());
+
+ assertThat(connected.get(8080), contains(serviceName));
+ assertThat(connected.get(8090), contains(serviceName));
+
+ messageSender.onDisconnected();
+ assertThat(connected.get(8080).isEmpty(), is(true));
+ assertThat(connected.get(8090).isEmpty(), is(true));
+ }
+
+ private void killServerReceivedMessage() {
+ int index = 0;
+ for (int port : eventsMap.keySet()) {
+ if (!eventsMap.get(port).isEmpty()) {
+ servers.get(index).shutdownNow();
+ }
+ index++;
+ }
+ }
+
+ private static class MyTxEventService extends TxEventServiceImplBase {
+ private final Set<String> connected;
+ private final Queue<TxEvent> events;
+
+ private MyTxEventService(Set<String> connected, Queue<TxEvent> events) {
+ this.connected = connected;
+ this.events = events;
+ }
+
+ @Override
+ public void onConnected(GrpcServiceConfig request, StreamObserver<GrpcCompensateCommand> responseObserver) {
+ connected.add(request.getInstanceId());
+ }
+
+ @Override
+ public void onTxEvent(GrpcTxEvent request, StreamObserver<GrpcAck> responseObserver) {
+ events.offer(new TxEvent(
+ request.getGlobalTxId(),
+ request.getLocalTxId(),
+ request.getParentTxId(),
+ request.getCompensationMethod(),
+ new String(request.getPayloads().toByteArray())));
+
+ responseObserver.onNext(GrpcAck.newBuilder().build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void onDisconnected(GrpcServiceConfig request, StreamObserver<GrpcAck> responseObserver) {
+ connected.remove(request.getInstanceId());
+ responseObserver.onNext(GrpcAck.newBuilder().build());
+ responseObserver.onCompleted();
+ }
+ }
+}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/resources/log4j2-test.xml b/omega/omega-connector/omega-connector-grpc/src/test/resources/log4j2-test.xml
new file mode 100644
index 0000000..58924c6
--- /dev/null
+++ b/omega/omega-connector/omega-connector-grpc/src/test/resources/log4j2-test.xml
@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<Configuration status="WARN">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="info">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>
diff --git a/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java b/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java
index 17674b7..d3c591e 100644
--- a/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java
+++ b/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java
@@ -37,7 +37,7 @@ public class MessageFormatTestBase {
@Test
public void serializeObjectIntoBytes() throws Exception {
- byte[] bytes = format.serialize(eventOf("hello", "world"));
+ byte[] bytes = format.serialize(new String[]{"hello", "world"});
Object[] message = format.deserialize(bytes);
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSerializer.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSerializer.java
index 7dfe0ca..0bc1e46 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSerializer.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSerializer.java
@@ -18,7 +18,5 @@
package org.apache.servicecomb.saga.omega.transaction;
public interface MessageSerializer {
- byte[] serialize(TxEvent event);
-
byte[] serialize(Object[] objects);
}
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.
[incubator-servicecomb-saga] 02/07: SCB-168 load balance based on
server latency
Posted by ni...@apache.org.
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 01c4a1d415576ad9920fa84cd81ad37dc4f6c637
Author: seanyinx <se...@huawei.com>
AuthorDate: Mon Jan 8 16:05:15 2018 +0800
SCB-168 load balance based on server latency
Signed-off-by: seanyinx <se...@huawei.com>
---
.../grpc/LoadBalancedClusterMessageSender.java | 128 +++++++++------------
.../grpc/LoadBalancedClusterMessageSenderTest.java | 84 +++++++++++---
.../saga/omega/spring/OmegaSpringConfig.java | 59 ++--------
.../saga/omega/transaction/MessageSender.java | 3 +
4 files changed, 141 insertions(+), 133 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
index 276f887..e5c9d19 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
@@ -17,13 +17,16 @@
package org.apache.servicecomb.saga.omega.connector.grpc;
+import static java.util.Collections.emptyList;
+
import java.lang.invoke.MethodHandles;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.net.URI;
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.function.Consumer;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
@@ -34,51 +37,61 @@ import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import io.grpc.Attributes;
-import io.grpc.EquivalentAddressGroup;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
-import io.grpc.NameResolver;
-import io.grpc.NameResolver.Factory;
-import io.grpc.util.RoundRobinLoadBalancerFactory;
public class LoadBalancedClusterMessageSender implements MessageSender {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final MessageSender messageSender;
+ private final Map<MessageSender, Long> senders = new HashMap<>();
+ private final Collection<ManagedChannel> channels;
- public LoadBalancedClusterMessageSender(String addresses,
+ public LoadBalancedClusterMessageSender(String[] addresses,
MessageSerializer serializer,
MessageDeserializer deserializer,
ServiceConfig serviceConfig,
MessageHandler handler) {
- this(new GrpcClientMessageSender(clusterDirectAddressChannel(addresses),
- serializer,
- deserializer,
- serviceConfig,
- handler));
- }
+ if (addresses.length == 0) {
+ throw new IllegalArgumentException("No reachable cluster address provided");
+ }
- LoadBalancedClusterMessageSender(MessageSender messageSender) {
- this.messageSender = messageSender;
+ channels = new ArrayList<>(addresses.length);
+ for (String address : addresses) {
+ ManagedChannel channel = ManagedChannelBuilder.forTarget(address)
+ .usePlaintext(true)
+ .build();
+
+ channels.add(channel);
+ senders.put(
+ new GrpcClientMessageSender(channel,
+ serializer,
+ deserializer,
+ serviceConfig,
+ handler),
+ 0L);
+ }
}
- private static ManagedChannel clusterDirectAddressChannel(String addresses) {
- return ManagedChannelBuilder.forTarget(addresses)
- .nameResolverFactory(new ClusterNameResolverFactory(addresses))
- .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
- .usePlaintext(true)
- .build();
+ LoadBalancedClusterMessageSender(MessageSender... messageSenders) {
+ for (MessageSender sender : messageSenders) {
+ senders.put(sender, 0L);
+ }
+ channels = emptyList();
}
@Override
public void onConnected() {
- messageSender.onConnected();
+ senders.keySet().forEach(MessageSender::onConnected);
}
@Override
public void onDisconnected() {
- messageSender.onDisconnected();
+ senders.keySet().forEach(MessageSender::onDisconnected);
+ }
+
+ @Override
+ public void close() {
+ channels.forEach(ManagedChannel::shutdownNow);
}
@Override
@@ -86,54 +99,27 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
boolean success = false;
do {
try {
- messageSender.send(event);
+ withFastestSender(messageSender -> {
+ // very large latency on exception
+ senders.put(messageSender, Long.MAX_VALUE);
+
+ 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);
}
} while (!success && !Thread.currentThread().isInterrupted());
-
}
- private static class ClusterNameResolverFactory extends Factory {
- private final String addresses;
-
- private ClusterNameResolverFactory(String addresses) {
- this.addresses = addresses;
- }
-
- @Override
- public NameResolver newNameResolver(URI targetUri, Attributes params) {
- return new NameResolver() {
- @Override
- public String getServiceAuthority() {
- return "localhost";
- }
-
- @Override
- public void start(final Listener listener) {
- List<SocketAddress> socketAddresses = Arrays.stream(addresses.split(","))
- .map(address -> {
- String[] split = address.split(":");
- return new InetSocketAddress(split[0], Integer.parseInt(split[1]));
- })
- .collect(Collectors.toList());
-
- listener.onAddresses(
- Arrays.asList(new EquivalentAddressGroup(socketAddresses.get(0)),
- new EquivalentAddressGroup(socketAddresses.get(1))),
- Attributes.EMPTY);
- }
-
- @Override
- public void shutdown() {
- }
- };
- }
-
- @Override
- public String getDefaultScheme() {
- return "directaddress";
- }
+ private void withFastestSender(Consumer<MessageSender> consumer) {
+ senders.entrySet()
+ .stream()
+ .min(Comparator.comparingLong(Entry::getValue))
+ .map(Entry::getKey)
+ .ifPresent(consumer);
}
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
index 4fee0e6..6f2d90b 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -17,10 +17,13 @@
package org.apache.servicecomb.saga.omega.connector.grpc;
+import static com.seanyinx.github.unit.scaffolding.AssertUtils.expectFailing;
import static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
@@ -48,16 +51,19 @@ import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTxEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceImplBase;
+import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
-import org.junit.Ignore;
+import org.junit.FixMethodOrder;
import org.junit.Test;
+import org.junit.runners.MethodSorters;
import org.mockito.Mockito;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class LoadBalancedClusterMessageSenderTest {
private static final int[] ports = {8080, 8090};
@@ -66,6 +72,11 @@ public class LoadBalancedClusterMessageSenderTest {
private static final Queue<TxEvent> eventsOn8080 = new ConcurrentLinkedQueue<>();
private static final Queue<TxEvent> eventsOn8090 = new ConcurrentLinkedQueue<>();
+ private static final Map<Integer, Integer> delays = new HashMap<Integer, Integer>() {{
+ put(8080, 0);
+ put(8090, 100);
+ }};
+
private static final Map<Integer, Set<String>> connected = new HashMap<Integer, Set<String>>() {{
put(8080, new ConcurrentSkipListSet<>());
put(8090, new ConcurrentSkipListSet<>());
@@ -76,8 +87,6 @@ public class LoadBalancedClusterMessageSenderTest {
put(8090, eventsOn8090);
}};
- private final String addresses = "localhost:8080,localhost:8090";
-
private final MessageSerializer serializer = objects -> objects[0].toString().getBytes();
private final MessageDeserializer deserializer = message -> new Object[] {new String(message)};
@@ -91,18 +100,23 @@ public class LoadBalancedClusterMessageSenderTest {
private final TxEvent event = new TxEvent(globalTxId, localTxId, parentTxId, compensationMethod, "blah");
private final String serviceName = uniquify("serviceName");
- private final MessageSender messageSender = new LoadBalancedClusterMessageSender(
- addresses,
- serializer,
- deserializer,
- new ServiceConfig(serviceName),
- handler);
+ private final String[] addresses = {"localhost:8080", "localhost:8090"};
+ private final MessageSender messageSender = newMessageSender(addresses);
+
+ private MessageSender newMessageSender(String[] addresses) {
+ return new LoadBalancedClusterMessageSender(
+ addresses,
+ serializer,
+ deserializer,
+ new ServiceConfig(serviceName),
+ handler);
+ }
@BeforeClass
public static void beforeClass() throws Exception {
Arrays.stream(ports).forEach(port -> {
ServerBuilder<?> serverBuilder = ServerBuilder.forPort(port);
- serverBuilder.addService(new MyTxEventService(connected.get(port), eventsMap.get(port)));
+ serverBuilder.addService(new MyTxEventService(connected.get(port), eventsMap.get(port), delays.get(port)));
Server server = serverBuilder.build();
try {
@@ -119,8 +133,14 @@ public class LoadBalancedClusterMessageSenderTest {
servers.forEach(Server::shutdown);
}
+ @After
+ public void after() throws Exception {
+ eventsOn8080.clear();
+ eventsOn8090.clear();
+ }
+
@Test
- public void reconnectOnConnectionLoss() throws Exception {
+ public void resendToAnotherServerOnFailure() throws Exception {
messageSender.send(event);
killServerReceivedMessage();
@@ -150,7 +170,6 @@ public class LoadBalancedClusterMessageSenderTest {
thread.join();
}
- @Ignore
@Test
public void broadcastConnectionAndDisconnection() throws Exception {
messageSender.onConnected();
@@ -164,6 +183,29 @@ public class LoadBalancedClusterMessageSenderTest {
assertThat(connected.get(8090).isEmpty(), is(true));
}
+ @Test
+ public void considerFasterServerFirst() throws Exception {
+ // we don't know which server is selected at first
+ messageSender.send(event);
+
+ // but now we only send to the one with lowest latency
+ messageSender.send(event);
+ messageSender.send(event);
+
+ assertThat(eventsOn8080.size(), greaterThanOrEqualTo(2));
+ assertThat(eventsOn8090.size(), lessThanOrEqualTo(1));
+ }
+
+ @Test
+ public void blowsUpWhenNoServerAddressProvided() throws Exception {
+ try {
+ newMessageSender(new String[0]);
+ expectFailing(IllegalArgumentException.class);
+ } catch (IllegalArgumentException e) {
+ assertThat(e.getMessage(), is("No reachable cluster address provided"));
+ }
+ }
+
private void killServerReceivedMessage() {
int index = 0;
for (int port : eventsMap.keySet()) {
@@ -177,15 +219,17 @@ public class LoadBalancedClusterMessageSenderTest {
private static class MyTxEventService extends TxEventServiceImplBase {
private final Set<String> connected;
private final Queue<TxEvent> events;
+ private final int delay;
- private MyTxEventService(Set<String> connected, Queue<TxEvent> events) {
+ private MyTxEventService(Set<String> connected, Queue<TxEvent> events, int delay) {
this.connected = connected;
this.events = events;
+ this.delay = delay;
}
@Override
public void onConnected(GrpcServiceConfig request, StreamObserver<GrpcCompensateCommand> responseObserver) {
- connected.add(request.getInstanceId());
+ connected.add(request.getServiceName());
}
@Override
@@ -197,13 +241,23 @@ public class LoadBalancedClusterMessageSenderTest {
request.getCompensationMethod(),
new String(request.getPayloads().toByteArray())));
+ sleep();
+
responseObserver.onNext(GrpcAck.newBuilder().build());
responseObserver.onCompleted();
}
+ private void sleep() {
+ try {
+ Thread.sleep(delay);
+ } catch (InterruptedException e) {
+ fail(e.getMessage());
+ }
+ }
+
@Override
public void onDisconnected(GrpcServiceConfig request, StreamObserver<GrpcAck> responseObserver) {
- connected.remove(request.getInstanceId());
+ connected.remove(request.getServiceName());
responseObserver.onNext(GrpcAck.newBuilder().build());
responseObserver.onCompleted();
}
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index 7ed1f84..00afc1a 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -17,14 +17,7 @@
package org.apache.servicecomb.saga.omega.spring;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import javax.annotation.PreDestroy;
-
-import org.apache.servicecomb.saga.omega.connector.grpc.GrpcClientMessageSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.LoadBalancedClusterMessageSender;
import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
@@ -32,22 +25,13 @@ import org.apache.servicecomb.saga.omega.context.UniqueIdGenerator;
import org.apache.servicecomb.saga.omega.format.KryoMessageFormat;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
-import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
-
@Configuration
class OmegaSpringConfig {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- private final List<ManagedChannel> channels = new ArrayList<>();
- private final List<MessageSender> senders = new ArrayList<>();
@Bean
IdGenerator<String> idGenerator() {
@@ -64,40 +48,21 @@ class OmegaSpringConfig {
return new ServiceConfig(serviceName);
}
- @PreDestroy
- void close() {
- senders.forEach(MessageSender::onDisconnected);
- channels.forEach(ManagedChannel::shutdown);
- }
-
@Bean
- MessageSender grpcMessageSender(@Value("${alpha.cluster.address}") String[] addresses, ServiceConfig serviceConfig,
+ MessageSender grpcMessageSender(
+ @Value("${alpha.cluster.address}") String[] addresses,
+ ServiceConfig serviceConfig,
@Lazy MessageHandler handler) {
- // TODO: 2017/12/26 connect to the one with lowest latency
- for (String address : addresses) {
- try {
- MessageSender sender = new GrpcClientMessageSender(grpcChannel(address), new KryoMessageFormat(),
- new KryoMessageFormat(), serviceConfig, handler);
- sender.onConnected();
- senders.add(sender);
- return sender;
- } catch (Exception e) {
- log.error("Unable to connect to alpha at {}", address, e);
- }
- }
-
- throw new IllegalArgumentException(
- "None of the alpha cluster is reachable: " + Arrays.toString(addresses));
- }
- private ManagedChannel grpcChannel(String address) {
- String[] pair = address.split(":");
+ MessageSender sender = new LoadBalancedClusterMessageSender(
+ addresses,
+ new KryoMessageFormat(),
+ new KryoMessageFormat(),
+ serviceConfig,
+ handler);
- ManagedChannel channel = ManagedChannelBuilder.forAddress(pair[0], Integer.parseInt(pair[1]))
- .usePlaintext(true)
- .build();
+ Runtime.getRuntime().addShutdownHook(new Thread(sender::close));
- channels.add(channel);
- return channel;
+ return sender;
}
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
index d37c5b8..39b4d62 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
@@ -24,5 +24,8 @@ public interface MessageSender {
default void onDisconnected() {
}
+ default void close() {
+ }
+
void send(TxEvent event);
}
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.
[incubator-servicecomb-saga] 05/07: SCB-168 resolved rebase conflict
Posted by ni...@apache.org.
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 e2828ee8aa781fcfa0e09a1c72ae2c88d8044dc7
Author: seanyinx <se...@huawei.com>
AuthorDate: Mon Jan 8 16:39:47 2018 +0800
SCB-168 resolved rebase conflict
Signed-off-by: seanyinx <se...@huawei.com>
---
.../servicecomb/saga/omega/format/MessageFormat.java | 12 +-----------
.../servicecomb/saga/omega/spring/OmegaSpringConfig.java | 16 +++++++++-------
2 files changed, 10 insertions(+), 18 deletions(-)
diff --git a/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/MessageFormat.java b/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/MessageFormat.java
index 0ccac37..d80467d 100644
--- a/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/MessageFormat.java
+++ b/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/MessageFormat.java
@@ -19,16 +19,6 @@ package org.apache.servicecomb.saga.omega.format;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageSerializer;
-import org.apache.servicecomb.saga.omega.transaction.OmegaException;
-import org.apache.servicecomb.saga.omega.transaction.TxEvent;
-interface MessageFormat extends MessageSerializer, MessageDeserializer {
- @Override
- default byte[] serialize(TxEvent event) {
- try {
- return serialize(event.payloads());
- } catch (OmegaException e) {
- throw new OmegaException("Unable to serialize event with global tx id " + event.globalTxId(), e);
- }
- }
+public interface MessageFormat extends MessageSerializer, MessageDeserializer {
}
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index 00afc1a..5c6a397 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -23,6 +23,7 @@ import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.context.UniqueIdGenerator;
import org.apache.servicecomb.saga.omega.format.KryoMessageFormat;
+import org.apache.servicecomb.saga.omega.format.MessageFormat;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.springframework.beans.factory.annotation.Value;
@@ -54,14 +55,15 @@ class OmegaSpringConfig {
ServiceConfig serviceConfig,
@Lazy MessageHandler handler) {
- MessageSender sender = new LoadBalancedClusterMessageSender(
- addresses,
- new KryoMessageFormat(),
- new KryoMessageFormat(),
- serviceConfig,
- handler);
+ MessageFormat messageFormat = new KryoMessageFormat();
+ MessageSender sender = new LoadBalancedClusterMessageSender(
+ addresses,
+ messageFormat,
+ messageFormat,
+ serviceConfig,
+ handler);
- Runtime.getRuntime().addShutdownHook(new Thread(sender::close));
+ Runtime.getRuntime().addShutdownHook(new Thread(sender::close));
return sender;
}
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.
[incubator-servicecomb-saga] 06/07: SCB-168 removed default logging
from spring boot
Posted by ni...@apache.org.
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 1e7ff02c8a8691758fcb74decded944e7125cc18
Author: seanyinx <se...@huawei.com>
AuthorDate: Mon Jan 8 16:45:12 2018 +0800
SCB-168 removed default logging from spring boot
Signed-off-by: seanyinx <se...@huawei.com>
---
omega/omega-connector/omega-connector-grpc/pom.xml | 5 ++++-
.../servicecomb/saga/omega/format/KryoMessageFormatTest.java | 6 +++---
.../servicecomb/saga/omega/format/MessageFormatTestBase.java | 9 ++-------
omega/omega-spring-tx/pom.xml | 6 ++++++
4 files changed, 15 insertions(+), 11 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/pom.xml b/omega/omega-connector/omega-connector-grpc/pom.xml
index 258eb4b..8ed9aa8 100644
--- a/omega/omega-connector/omega-connector-grpc/pom.xml
+++ b/omega/omega-connector/omega-connector-grpc/pom.xml
@@ -53,19 +53,22 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
+ <scope>test</scope>
</dependency>
-
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
diff --git a/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormatTest.java b/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormatTest.java
index 1f63d74..327b6ec 100644
--- a/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormatTest.java
+++ b/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormatTest.java
@@ -17,7 +17,7 @@
package org.apache.servicecomb.saga.omega.format;
-import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertThat;
import org.junit.BeforeClass;
@@ -32,10 +32,10 @@ public class KryoMessageFormatTest extends MessageFormatTestBase {
@Test
public void serializeEmptyClassIntoBytes() {
- byte[] bytes = format.serialize(eventOf(new EmptyClass()));
+ byte[] bytes = format.serialize(new Object[]{new EmptyClass()});
Object[] message = format.deserialize(bytes);
- assertThat(message[0] instanceof EmptyClass, is(true));
+ assertThat(message[0], instanceOf(EmptyClass.class));
}
}
diff --git a/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java b/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java
index d3c591e..aab6ad7 100644
--- a/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java
+++ b/omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/MessageFormatTestBase.java
@@ -26,7 +26,6 @@ import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertThat;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
-import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.junit.Ignore;
import org.junit.Test;
@@ -37,7 +36,7 @@ public class MessageFormatTestBase {
@Test
public void serializeObjectIntoBytes() throws Exception {
- byte[] bytes = format.serialize(new String[]{"hello", "world"});
+ byte[] bytes = format.serialize(new Object[]{"hello", "world"});
Object[] message = format.deserialize(bytes);
@@ -46,7 +45,7 @@ public class MessageFormatTestBase {
@Test
public void serializeNullIntoBytes() throws Exception {
- byte[] bytes = format.serialize(eventOf((Object[]) null));
+ byte[] bytes = format.serialize(null);
Object[] message = format.deserialize(bytes);
@@ -63,10 +62,6 @@ public class MessageFormatTestBase {
}
}
- TxEvent eventOf(Object... payloads) {
- return new TxEvent(null, null, null, null, payloads);
- }
-
static class EmptyClass {
}
}
diff --git a/omega/omega-spring-tx/pom.xml b/omega/omega-spring-tx/pom.xml
index 991ad93..539335b 100644
--- a/omega/omega-spring-tx/pom.xml
+++ b/omega/omega-spring-tx/pom.xml
@@ -32,6 +32,12 @@
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.servicecomb.saga</groupId>
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.
[incubator-servicecomb-saga] 07/07: SCB-168 connected to alpha on
start
Posted by ni...@apache.org.
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 367b1cc60c26b8bb30e42c6013de212f0e8568e9
Author: seanyinx <se...@huawei.com>
AuthorDate: Mon Jan 8 17:14:35 2018 +0800
SCB-168 connected to alpha on start
Signed-off-by: seanyinx <se...@huawei.com>
---
.../org/apache/servicecomb/saga/integration/pack/tests/PackIT.java | 4 ++--
.../org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java | 6 +++++-
2 files changed, 7 insertions(+), 3 deletions(-)
diff --git a/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/saga/integration/pack/tests/PackIT.java b/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/saga/integration/pack/tests/PackIT.java
index 32d4254..56092be 100644
--- a/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/saga/integration/pack/tests/PackIT.java
+++ b/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/saga/integration/pack/tests/PackIT.java
@@ -67,7 +67,7 @@ public class PackIT {
repository.deleteAll();
}
- @Test
+ @Test(timeout = 5000)
public void updatesTxStateToAlpha() throws Exception {
ResponseEntity<String> entity = restTemplate.getForEntity("/greet?name={name}",
String.class,
@@ -129,7 +129,7 @@ public class PackIT {
assertThat(compensatedMessages.isEmpty(), is(true));
}
- @Test
+ @Test(timeout = 5000)
public void compensatesFailedGlobalTransaction() throws Exception {
ResponseEntity<String> entity = restTemplate.getForEntity("/greet?name={name}",
String.class,
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index 5c6a397..b5d118a 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -63,7 +63,11 @@ class OmegaSpringConfig {
serviceConfig,
handler);
- Runtime.getRuntime().addShutdownHook(new Thread(sender::close));
+ sender.onConnected();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ sender.onDisconnected();
+ sender.close();
+ }));
return sender;
}
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.
[incubator-servicecomb-saga] 04/07: SCB-168 made latency
significant enough to pass test
Posted by ni...@apache.org.
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 55296894828597f90140707e988e050f85f55d6b
Author: seanyinx <se...@huawei.com>
AuthorDate: Mon Jan 8 16:25:36 2018 +0800
SCB-168 made latency significant enough to pass test
Signed-off-by: seanyinx <se...@huawei.com>
---
.../connector/grpc/LoadBalancedClusterMessageSenderTest.java | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
index dc75663..1868f31 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -22,8 +22,6 @@ import static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
@@ -75,7 +73,7 @@ public class LoadBalancedClusterMessageSenderTest {
private static final Map<Integer, Integer> delays = new HashMap<Integer, Integer>() {{
put(8080, 0);
- put(8090, 100);
+ put(8090, 500);
}};
private static final Map<Integer, Set<String>> connected = new HashMap<Integer, Set<String>>() {{
@@ -222,9 +220,10 @@ public class LoadBalancedClusterMessageSenderTest {
// but now we only send to the one with lowest latency
messageSender.send(event);
messageSender.send(event);
+ messageSender.send(event);
- assertThat(eventsOn8080.size(), greaterThanOrEqualTo(2));
- assertThat(eventsOn8090.size(), lessThanOrEqualTo(1));
+ assertThat(eventsOn8080.size(), is(3));
+ assertThat(eventsOn8090.size(), is(1));
}
@Test
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.