You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by se...@apache.org on 2018/01/08 08:39:59 UTC
[incubator-servicecomb-saga] branch SCB-168_load_balance updated
(bc064fe -> 65c94f9)
This is an automated email from the ASF dual-hosted git repository.
seanyinx pushed a change to branch SCB-168_load_balance
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git.
discard bc064fe SCB-168 made latency significant enough to pass test
discard e25d9a2 SCB-168 exception free connection & disconnection
discard 8a8d61a SCB-168 load balance based on server latency
discard c9a8f39 SCB-168 load balanced alpha cluster client
add 828b2a7 SCB-174 add kryo serializer
add d726294 SCB-174 remove redudant test codes
add 390271d SCB-174 retrieve kryo instance from pool
add 6d4c084 SCB-174 remove useless NativeMessageFormat implementation
new c731fbb SCB-168 load balanced alpha cluster client
new 238aad2 SCB-168 load balance based on server latency
new cb189d8 SCB-168 exception free connection & disconnection
new 1dd94ac SCB-168 made latency significant enough to pass test
new 65c94f9 SCB-168 resolved rebase conflict
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (bc064fe)
\
N -- N -- N refs/heads/SCB-168_load_balance (65c94f9)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 5 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:
omega/omega-format/pom.xml | 4 ++
...veMessageFormat.java => KryoMessageFormat.java} | 49 +++++++++++++---------
.../saga/omega/format/MessageFormat.java | 9 ++--
.../saga/omega/format/KryoMessageFormatTest.java | 23 ++++++----
...eFormatTest.java => MessageFormatTestBase.java} | 25 ++++++-----
.../saga/omega/spring/OmegaSpringConfig.java | 8 ++--
pom.xml | 6 +++
7 files changed, 77 insertions(+), 47 deletions(-)
rename omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/{NativeMessageFormat.java => KryoMessageFormat.java} (54%)
copy alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/OmegaCallback.java => omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/MessageFormat.java (74%)
copy saga-core/src/test/java/org/apache/servicecomb/saga/core/LongIdGeneratorTest.java => omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormatTest.java (64%)
rename omega/omega-format/src/test/java/org/apache/servicecomb/saga/omega/format/{NativeMessageFormatTest.java => MessageFormatTestBase.java} (79%)
--
To stop receiving notification emails like this one, please contact
['"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>'].
[incubator-servicecomb-saga] 03/05: SCB-168 exception free
connection & disconnection
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
seanyinx pushed a commit to branch SCB-168_load_balance
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit cb189d8519b17e734e9ff19fe540a11038f7578f
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 b2e837c..58a1bd1 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] 05/05: SCB-168 resolved rebase conflict
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
seanyinx pushed a commit to branch SCB-168_load_balance
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 65c94f9667677796ceb0315bf85d96e66ac29b14
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] 01/05: SCB-168 load balanced alpha
cluster client
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
seanyinx pushed a commit to branch SCB-168_load_balance
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit c731fbb2438c02af94e9277537442196f870ff1d
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 18241a9..ba001c9 100644
--- a/omega/omega-connector/omega-connector-grpc/pom.xml
+++ b/omega/omega-connector/omega-connector-grpc/pom.xml
@@ -52,6 +52,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>
@@ -69,5 +81,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/05: SCB-168 load balance based on
server latency
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
seanyinx pushed a commit to branch SCB-168_load_balance
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 238aad27b1546da456119936ddf903f2e2d1706b
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] 04/05: SCB-168 made latency
significant enough to pass test
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
seanyinx pushed a commit to branch SCB-168_load_balance
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 1dd94ac412b99c3b522e341ebe88f9dd5d750358
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>.