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/07/31 08:38:21 UTC
[incubator-servicecomb-saga] branch master updated (b1c3fb9 ->
3f5c38a)
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 b1c3fb9 SCB-692 Fixed the build error with Spring-Boot-2
new 316af59 Updated the gitter room address
new 8667889 SCB-648 Updated the code of omega-context, omega-transaction
new f20f144 SCB-648 Update the omega-transport modules
new 1f19364 SCB-648 Update the omega-format modules
new 239acca SCB-648 Added the omega-connector-grpc module
new 3f5c38a SCB-648 Added the omega-spring-starter
The 6 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:
README.md | 2 +-
README_ZH.md | 2 +-
.../connector/grpc/GrpcClientMessageSender.java | 12 ++-
.../grpc/LoadBalancedClusterMessageSender.java | 73 ++++++++++-----
.../connector/grpc/RetryableMessageSender.java | 20 +++++
...LoadBalanceClusterMessageSenderWithTLSTest.java | 13 ++-
.../grpc/LoadBalancedClusterMessageSenderTest.java | 87 ++++++++++++++----
.../LoadBalancedClusterMessageSenderTestBase.java | 46 +++++++---
.../grpc/PushBackReconnectRunnableTest.java | 11 ++-
.../connector/grpc/RetryableMessageSenderTest.java | 17 ++--
omega/omega-context/pom.xml | 5 ++
.../saga/omega/context/OmegaContextTest.java | 83 +++++++++++------
.../saga/omega/context/UniqueIdGeneratorTest.java | 7 +-
.../saga/omega/format/KryoMessageFormat.java | 7 +-
.../saga/omega/format/MessageFormatTestBase.java | 5 +-
.../saga/omega/spring/OmegaSpringConfig.java | 13 +--
.../transaction/spring/ExecutorFieldCallback.java | 12 ++-
.../spring/TransactionInterceptionTest.java | 100 +++++++++++++++++----
.../saga/omega/transaction/MessageSender.java | 15 +---
.../transaction/CompensableInterceptorTest.java | 30 +++++--
.../CompensationMessageHandlerTest.java | 28 +++++-
.../omega/transaction/DefaultRecoveryTest.java | 35 ++++++--
.../omega/transaction/ForwardRecoveryTest.java | 54 ++++++++---
.../SagaStartAnnotationProcessorTest.java | 31 ++++++-
.../omega/transaction/SagaStartAspectTest.java | 26 +++++-
.../omega/transaction/TransactionAspectTest.java | 26 +++++-
.../src/test/java/SagaDubboConsumerFilterTest.java | 9 +-
.../src/test/java/SagaDubboProviderFilterTest.java | 8 +-
.../feign/FeignAutoConfigurationTest.java | 19 +++-
.../TransactionHandlerInterceptorTest.java | 9 +-
.../servicecomb/SagaConsumerHandlerTest.java | 8 +-
.../servicecomb/SagaProviderHandlerTest.java | 12 ++-
omega/pom.xml | 16 ++++
33 files changed, 658 insertions(+), 183 deletions(-)
[incubator-servicecomb-saga] 02/06: SCB-648 Updated the code of
omega-context, omega-transaction
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 866788903c94b63811f97e4e4e441f47e2e3728a
Author: Willem Jiang <ji...@huawei.com>
AuthorDate: Mon Jul 30 16:55:08 2018 +0800
SCB-648 Updated the code of omega-context, omega-transaction
---
omega/omega-context/pom.xml | 5 ++
.../saga/omega/context/OmegaContextTest.java | 83 +++++++++++++++-------
.../saga/omega/context/UniqueIdGeneratorTest.java | 7 +-
.../transaction/spring/ExecutorFieldCallback.java | 12 ++--
.../spring/TransactionInterceptionTest.java | 14 +++-
.../saga/omega/transaction/MessageSender.java | 15 ++--
.../transaction/CompensableInterceptorTest.java | 30 ++++++--
.../CompensationMessageHandlerTest.java | 28 +++++++-
.../omega/transaction/DefaultRecoveryTest.java | 35 +++++++--
.../omega/transaction/ForwardRecoveryTest.java | 54 ++++++++++----
.../SagaStartAnnotationProcessorTest.java | 31 ++++++--
.../omega/transaction/SagaStartAspectTest.java | 26 ++++++-
.../omega/transaction/TransactionAspectTest.java | 26 ++++++-
omega/pom.xml | 16 +++++
14 files changed, 299 insertions(+), 83 deletions(-)
diff --git a/omega/omega-context/pom.xml b/omega/omega-context/pom.xml
index 4f9e38a..654295c 100644
--- a/omega/omega-context/pom.xml
+++ b/omega/omega-context/pom.xml
@@ -42,6 +42,11 @@
<groupId>com.github.seanyinx</groupId>
<artifactId>unit-scaffolding</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-common</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
diff --git a/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/OmegaContextTest.java b/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/OmegaContextTest.java
index b741e88..a104f7e 100644
--- a/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/OmegaContextTest.java
+++ b/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/OmegaContextTest.java
@@ -21,62 +21,91 @@ import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
+import java.io.Serializable;
import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
+
import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import org.junit.Test;
+
public class OmegaContextTest {
- private final OmegaContext omegaContext = new OmegaContext(() -> "ignored");
+ private final OmegaContext omegaContext = new OmegaContext(
+ new IdGenerator() {
+ @Override
+ public Serializable nextId() {
+ return "ignored";
+ }
+ });
+
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+
@Test
public void eachThreadGetsDifferentGlobalTxId() throws Exception {
- CyclicBarrier barrier = new CyclicBarrier(2);
+ final CyclicBarrier barrier = new CyclicBarrier(2);
+
+ Runnable runnable = exceptionalRunnable(new ExceptionalRunnable() {
- Runnable runnable = exceptionalRunnable(() -> {
- String txId = UUID.randomUUID().toString();
- omegaContext.setGlobalTxId(txId);
- barrier.await();
+ @Override
+ public void run() throws Exception {
+ String txId = UUID.randomUUID().toString();
+ omegaContext.setGlobalTxId(txId);
+ barrier.await();
- assertThat(omegaContext.globalTxId(), is(txId));
+ assertThat(omegaContext.globalTxId(), is(txId));
+ }
});
- CompletableFuture<Void> future1 = CompletableFuture.runAsync(runnable);
- CompletableFuture<Void> future2 = CompletableFuture.runAsync(runnable);
+ Future f1 = executor.submit(runnable); ;
+ Future f2 = executor.submit(runnable);
+ f1.get();
+ f2.get();
- CompletableFuture.allOf(future1, future2).join();
}
@Test
public void eachThreadGetsDifferentLocalTxId() throws Exception {
- CyclicBarrier barrier = new CyclicBarrier(2);
+ final CyclicBarrier barrier = new CyclicBarrier(2);
- Runnable runnable = exceptionalRunnable(() -> {
- String spanId = UUID.randomUUID().toString();
- omegaContext.setLocalTxId(spanId);
- barrier.await();
+ Runnable runnable = exceptionalRunnable(new ExceptionalRunnable() {
- assertThat(omegaContext.localTxId(), is(spanId));
- });
+ @Override
+ public void run() throws Exception {
+ String spanId = UUID.randomUUID().toString();
+ omegaContext.setLocalTxId(spanId);
+ barrier.await();
- CompletableFuture<Void> future1 = CompletableFuture.runAsync(runnable);
- CompletableFuture<Void> future2 = CompletableFuture.runAsync(runnable);
+ assertThat(omegaContext.localTxId(), is(spanId));
+ }
+ });
- CompletableFuture.allOf(future1, future2).join();
+ Future f1 = executor.submit(runnable); ;
+ Future f2 = executor.submit(runnable);
+ f1.get();
+ f2.get();
}
- private Runnable exceptionalRunnable(ExceptionalRunnable runnable) {
- return () -> {
- try {
- runnable.run();
- } catch (Exception e) {
- fail(e.getMessage());
+ private Runnable exceptionalRunnable(final ExceptionalRunnable runnable) {
+ return new Runnable() {
+
+ @Override
+ public void run() {
+ try {
+ runnable.run();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
}
};
}
+
interface ExceptionalRunnable {
void run() throws Exception;
}
diff --git a/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/UniqueIdGeneratorTest.java b/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/UniqueIdGeneratorTest.java
index bcedcfa..7325a1e 100644
--- a/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/UniqueIdGeneratorTest.java
+++ b/omega/omega-context/src/test/java/org/apache/servicecomb/saga/omega/context/UniqueIdGeneratorTest.java
@@ -37,7 +37,12 @@ public class UniqueIdGeneratorTest {
private final UniqueIdGenerator idGenerator = new UniqueIdGenerator();
- private Callable<String> task = idGenerator::nextId;
+ private Callable<String> task = new Callable<String>() {
+ @Override
+ public String call() throws Exception {
+ return idGenerator.nextId();
+ }
+ };
@Test
public void nextIdIsUnique() throws InterruptedException {
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java
index e07356c..9ef937e 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java
+++ b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java
@@ -22,7 +22,9 @@ import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;
@@ -137,10 +139,12 @@ class ExecutorFieldCallback implements FieldCallback {
if (isExecutable(arg)) {
augmentedArgs[i] = RunnableProxy.newInstance(arg, omegaContext);
} else if (isCollectionOfExecutables(arg)) {
- augmentedArgs[i] = ((Collection<?>) arg)
- .stream()
- .map(a -> RunnableProxy.newInstance(a, omegaContext))
- .collect(Collectors.toList());
+ List argList = new ArrayList();
+ Collection argCollection = (Collection<?>) arg;
+ for (Object a : argCollection) {
+ argList.add(RunnableProxy.newInstance(a, omegaContext));
+ }
+ augmentedArgs[i] = argList;
} else {
augmentedArgs[i] = arg;
}
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
index 6505bfc..cc43865 100644
--- a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
@@ -254,11 +254,21 @@ public class TransactionInterceptionTest {
@Test
public void passesOmegaContextThroughDifferentThreads() throws Exception {
- new Thread(() -> userService.add(user)).start();
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ userService.add(user))
+ }
+ }).start();
waitTillSavedUser(username);
String localTxId = omegaContext.newLocalTxId();
- new Thread(() -> userService.add(jack)).start();
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ userService.add(jack);
+ }
+ }).start();
waitTillSavedUser(usernameJack);
assertArrayEquals(
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 5271b82..019d8dd 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
@@ -18,20 +18,13 @@
package org.apache.servicecomb.saga.omega.transaction;
public interface MessageSender {
- MessageSender NO_OP_SENDER = event -> new AlphaResponse(false);
+ void onConnected();
- default void onConnected() {
- }
+ void onDisconnected();
- default void onDisconnected() {
- }
+ void close();
- default void close() {
- }
-
- default String target() {
- return "UNKNOWN";
- }
+ String target();
AlphaResponse send(TxEvent event);
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java
index 1e01ea5..aefc9a4 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java
@@ -42,11 +42,31 @@ public class CompensableInterceptorTest {
private final String localTxId = UUID.randomUUID().toString();
private final String parentTxId = UUID.randomUUID().toString();
- private final MessageSender sender = e -> {
- messages.add(e);
- return new AlphaResponse(false);
+ private final MessageSender sender =new MessageSender() {
+ @Override
+ public void onConnected() {
+ }
+
+ @Override
+ public void onDisconnected() {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event);
+ return new AlphaResponse(false);
+ }
};
-
+
private final String message = uniquify("message");
private final String retryMethod = uniquify("retryMethod");
@@ -77,7 +97,7 @@ public class CompensableInterceptorTest {
assertThat(event.retryMethod(), is(retryMethod));
assertThat(event.type(), is(EventType.TxStartedEvent));
assertThat(event.compensationMethod(), is(compensationMethod));
- assertThat(asList(event.payloads()), contains(message));
+ assertThat(asList(event.payloads()).contains(message), is(true));
}
@Test
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java
index d5d5de5..ab1f988 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java
@@ -34,9 +34,31 @@ import org.junit.Test;
public class CompensationMessageHandlerTest {
private final List<TxEvent> events = new ArrayList<>();
- private final MessageSender sender = e -> {
- events.add(e);
- return new AlphaResponse(false);
+ private final MessageSender sender = new MessageSender() {
+ @Override
+ public void onConnected() {
+
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN"; }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ events.add(event);
+ return new AlphaResponse(false);
+ }
};
private final String globalTxId = uniquify("globalTxId");
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java
index 75062bc..edbabb9 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java
@@ -67,9 +67,32 @@ public class DefaultRecoveryTest {
private final Compensable compensable = mock(Compensable.class);
- private final MessageSender sender = e -> {
- messages.add(e);
- return new AlphaResponse(false);
+ private final MessageSender sender = new MessageSender() {
+ @Override
+ public void onConnected() {
+
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event);
+ return new AlphaResponse(false);
+ }
};
private final CompensableInterceptor interceptor = new CompensableInterceptor(omegaContext, sender);
@@ -143,7 +166,7 @@ public class DefaultRecoveryTest {
@Test
public void returnImmediatelyWhenReceivedRejectResponse() {
MessageSender sender = mock(MessageSender.class);
- when(sender.send(any())).thenReturn(new AlphaResponse(true));
+ when(sender.send(any(TxEvent.class))).thenReturn(new AlphaResponse(true));
CompensableInterceptor interceptor = new CompensableInterceptor(omegaContext, sender);
@@ -156,7 +179,7 @@ public class DefaultRecoveryTest {
fail("unexpected exception throw: " + throwable);
}
- verify(sender, times(1)).send(any());
+ verify(sender, times(1)).send(any(TxEvent.class));
}
@Test
@@ -179,4 +202,4 @@ public class DefaultRecoveryTest {
private String doNothing() {
return "doNothing";
}
-}
\ No newline at end of file
+}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java
index 76fe55a..9ba4b47 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java
@@ -66,9 +66,32 @@ public class ForwardRecoveryTest {
private final Compensable compensable = mock(Compensable.class);
- private final MessageSender sender = e -> {
- messages.add(e);
- return new AlphaResponse(false);
+ private final MessageSender sender = new MessageSender() {
+ @Override
+ public void onConnected() {
+
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event);
+ return new AlphaResponse(false);
+ }
};
private final CompensableInterceptor interceptor = new CompensableInterceptor(omegaContext, sender);
@@ -94,7 +117,7 @@ public class ForwardRecoveryTest {
@Test
public void forwardExceptionWhenGlobalTxAborted() {
MessageSender sender = mock(MessageSender.class);
- when(sender.send(any())).thenReturn(new AlphaResponse(true));
+ when(sender.send(any(TxEvent.class))).thenReturn(new AlphaResponse(true));
CompensableInterceptor interceptor = new CompensableInterceptor(omegaContext, sender);
@@ -107,7 +130,7 @@ public class ForwardRecoveryTest {
fail("unexpected exception throw: " + throwable);
}
- verify(sender, times(1)).send(any());
+ verify(sender, times(1)).send(any(TxEvent.class));
}
@Test
@@ -135,14 +158,17 @@ public class ForwardRecoveryTest {
when(compensable.retryDelayInMilliseconds()).thenReturn(1000);
when(joinPoint.proceed()).thenThrow(oops);
- Thread thread = new Thread(() -> {
- try {
- recoveryPolicy.apply(joinPoint, compensable, interceptor, omegaContext, parentTxId, -1);
- expectFailing(OmegaException.class);
- } catch (OmegaException e) {
- exception = e;
- } catch (Throwable throwable) {
- fail("unexpected exception throw: " + throwable);
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ recoveryPolicy.apply(joinPoint, compensable, interceptor, omegaContext, parentTxId, -1);
+ expectFailing(OmegaException.class);
+ } catch (OmegaException e) {
+ exception = e;
+ } catch (Throwable throwable) {
+ fail("unexpected exception throw: " + throwable);
+ }
}
});
thread.start();
@@ -156,4 +182,4 @@ public class ForwardRecoveryTest {
private String doNothing() {
return "doNothing";
}
-}
\ No newline at end of file
+}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java
index 8c496f6..39fcf5f 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java
@@ -42,9 +42,32 @@ public class SagaStartAnnotationProcessorTest {
private final List<TxEvent> messages = new ArrayList<>();
- private final MessageSender sender = e -> {
- messages.add(e);
- return new AlphaResponse(false);
+ private final MessageSender sender = new MessageSender() {
+ @Override
+ public void onConnected() {
+
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event);
+ return new AlphaResponse(false);
+ }
};
private final String globalTxId = UUID.randomUUID().toString();
@@ -96,7 +119,7 @@ public class SagaStartAnnotationProcessorTest {
MessageSender sender = mock(MessageSender.class);
SagaStartAnnotationProcessor sagaStartAnnotationProcessor = new SagaStartAnnotationProcessor(context, sender);
- doThrow(exception).when(sender).send(any());
+ doThrow(exception).when(sender).send(any(TxEvent.class));
try {
sagaStartAnnotationProcessor.preIntercept(null, null, 0, null, 0);
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java
index 77d40ef..10d55ff 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java
@@ -42,9 +42,29 @@ public class SagaStartAspectTest {
private final List<TxEvent> messages = new ArrayList<>();
private final String globalTxId = UUID.randomUUID().toString();
- private final MessageSender sender = e -> {
- messages.add(e);
- return new AlphaResponse(false);
+ private final MessageSender sender = new MessageSender() {
+ @Override
+ public void onConnected() {
+ }
+
+ @Override
+ public void onDisconnected() {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event);
+ return new AlphaResponse(false);
+ }
};
private final ProceedingJoinPoint joinPoint = Mockito.mock(ProceedingJoinPoint.class);
private final MethodSignature methodSignature = Mockito.mock(MethodSignature.class);
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java
index 0aa9549..d7fade6 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java
@@ -42,9 +42,29 @@ public class TransactionAspectTest {
private final String localTxId = UUID.randomUUID().toString();
private final String newLocalTxId = UUID.randomUUID().toString();
- private final MessageSender sender = e -> {
- messages.add(e);
- return new AlphaResponse(false);
+ private final MessageSender sender = new MessageSender() {
+ @Override
+ public void onConnected() {
+ }
+
+ @Override
+ public void onDisconnected() {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event);
+ return new AlphaResponse(false);
+ }
};
private final ProceedingJoinPoint joinPoint = mock(ProceedingJoinPoint.class);
private final MethodSignature methodSignature = mock(MethodSignature.class);
diff --git a/omega/pom.xml b/omega/pom.xml
index 2e2d3a8..bc431b5 100644
--- a/omega/pom.xml
+++ b/omega/pom.xml
@@ -74,4 +74,20 @@
</dependencies>
</dependencyManagement>
+ <build>
+ <plugins>
+ <!-- SCB-648 Just make sure the Omega is 1.7 compatible -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version>
+ <configuration>
+ <encoding>UTF-8</encoding>
+ <source>1.7</source>
+ <target>1.7</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
[incubator-servicecomb-saga] 06/06: SCB-648 Added the
omega-spring-starter
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 3f5c38a146982a04ed2b17759309de3e14f8c5ab
Author: Willem Jiang <ji...@huawei.com>
AuthorDate: Tue Jul 31 15:35:54 2018 +0800
SCB-648 Added the omega-spring-starter
---
.../servicecomb/saga/omega/spring/OmegaSpringConfig.java | 13 ++++++++-----
1 file changed, 8 insertions(+), 5 deletions(-)
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 aaf67f6..afa6443 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
@@ -76,7 +76,7 @@ class OmegaSpringConfig {
MessageFormat messageFormat = new KryoMessageFormat();
AlphaClusterConfig clusterConfig = new AlphaClusterConfig(Arrays.asList(addresses),
enableSSL, mutualAuth, cert, key, certChain);
- MessageSender sender = new LoadBalancedClusterMessageSender(
+ final MessageSender sender = new LoadBalancedClusterMessageSender(
clusterConfig,
messageFormat,
messageFormat,
@@ -85,11 +85,14 @@ class OmegaSpringConfig {
reconnectDelay);
sender.onConnected();
- Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- sender.onDisconnected();
- sender.close();
+
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ sender.onDisconnected();
+ sender.close();
+ }
}));
-
return sender;
}
}
[incubator-servicecomb-saga] 04/06: SCB-648 Update the omega-format
modules
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 1f19364309a9d7221d7eaef35e10e19e657434a6
Author: Willem Jiang <ji...@huawei.com>
AuthorDate: Tue Jul 31 15:16:09 2018 +0800
SCB-648 Update the omega-format modules
---
.../apache/servicecomb/saga/omega/format/KryoMessageFormat.java | 7 ++++++-
.../servicecomb/saga/omega/format/MessageFormatTestBase.java | 5 ++++-
2 files changed, 10 insertions(+), 2 deletions(-)
diff --git a/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormat.java b/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormat.java
index f948963..50dfba8 100644
--- a/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormat.java
+++ b/omega/omega-format/src/main/java/org/apache/servicecomb/saga/omega/format/KryoMessageFormat.java
@@ -32,7 +32,12 @@ public class KryoMessageFormat implements MessageFormat {
private static final int DEFAULT_BUFFER_SIZE = 4096;
- private static final KryoFactory factory = Kryo::new;
+ private static final KryoFactory factory = new KryoFactory() {
+ @Override
+ public Kryo create() {
+ return new Kryo();
+ }
+ };
private static final KryoPool pool = new KryoPool.Builder(factory).softReferences().build();
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 aab6ad7..904bd4c 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
@@ -25,7 +25,10 @@ import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertThat;
+import java.util.List;
+
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
+import org.hamcrest.Matcher;
import org.junit.Ignore;
import org.junit.Test;
@@ -40,7 +43,7 @@ public class MessageFormatTestBase {
Object[] message = format.deserialize(bytes);
- assertThat(asList(message), contains("hello", "world"));
+ assertThat(asList(message).containsAll(asList("hello", "world")), is(true));
}
@Test
[incubator-servicecomb-saga] 05/06: SCB-648 Added the
omega-connector-grpc module
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 239accaa8005e73a93f26cfdae96b066065176c4
Author: Willem Jiang <ji...@huawei.com>
AuthorDate: Tue Jul 31 15:35:20 2018 +0800
SCB-648 Added the omega-connector-grpc module
---
.../connector/grpc/GrpcClientMessageSender.java | 12 ++-
.../grpc/LoadBalancedClusterMessageSender.java | 73 ++++++++++++------
.../connector/grpc/RetryableMessageSender.java | 20 +++++
...LoadBalanceClusterMessageSenderWithTLSTest.java | 13 +++-
.../grpc/LoadBalancedClusterMessageSenderTest.java | 87 ++++++++++++++++-----
.../LoadBalancedClusterMessageSenderTestBase.java | 46 ++++++++---
.../grpc/PushBackReconnectRunnableTest.java | 11 ++-
.../connector/grpc/RetryableMessageSenderTest.java | 17 +++--
.../spring/TransactionInterceptionTest.java | 88 ++++++++++++++++++----
9 files changed, 285 insertions(+), 82 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 b33eeb3..13da6cf 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
@@ -20,8 +20,7 @@
package org.apache.servicecomb.saga.omega.connector.grpc;
-import java.util.function.Function;
-
+import org.apache.servicecomb.saga.omega.connector.grpc.LoadBalancedClusterMessageSender.ErrorHandlerFactory;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
@@ -58,7 +57,7 @@ public class GrpcClientMessageSender implements MessageSender {
MessageSerializer serializer,
MessageDeserializer deserializer,
ServiceConfig serviceConfig,
- Function<MessageSender, Runnable> errorHandlerFactory,
+ ErrorHandlerFactory errorHandlerFactory,
MessageHandler handler) {
this.target = address;
this.asyncEventService = TxEventServiceGrpc.newStub(channel);
@@ -66,7 +65,7 @@ public class GrpcClientMessageSender implements MessageSender {
this.serializer = serializer;
this.compensateStreamObserver =
- new GrpcCompensateStreamObserver(handler, errorHandlerFactory.apply(this), deserializer);
+ new GrpcCompensateStreamObserver(handler, errorHandlerFactory.getHandler(this), deserializer);
this.serviceConfig = serviceConfig(serviceConfig.serviceName(), serviceConfig.instanceId());
}
@@ -81,6 +80,11 @@ public class GrpcClientMessageSender implements MessageSender {
}
@Override
+ public void close() {
+ // just do nothing here
+ }
+
+ @Override
public String target() {
return target;
}
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 87c2a64..032601f 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
@@ -35,7 +35,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.function.Function;
import javax.net.ssl.SSLException;
@@ -110,7 +109,7 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
serializer,
deserializer,
serviceConfig,
- errorHandlerFactory(),
+ new ErrorHandlerFactory(),
handler),
0L);
}
@@ -128,30 +127,37 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
@Override
public void onConnected() {
- senders.keySet().forEach(sender -> {
+ for(MessageSender sender :senders.keySet()){
try {
sender.onConnected();
} catch (Exception e) {
LOG.error("Failed connecting to alpha at {}", sender.target(), e);
}
- });
+ };
}
@Override
public void onDisconnected() {
- senders.keySet().forEach(sender -> {
+ for (MessageSender sender :senders.keySet()) {
try {
sender.onDisconnected();
} catch (Exception e) {
LOG.error("Failed disconnecting from alpha at {}", sender.target(), e);
}
- });
+ };
}
@Override
public void close() {
scheduler.shutdown();
- channels.forEach(ManagedChannel::shutdownNow);
+ for(ManagedChannel channel : channels) {
+ channel.shutdownNow();
+ }
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
}
@Override
@@ -179,29 +185,50 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
}
private MessageSender fastestSender() {
- return senders.entrySet()
- .stream()
- .filter(entry -> entry.getValue() < Long.MAX_VALUE)
- .min(Comparator.comparingLong(Entry::getValue))
- .map(Entry::getKey)
- .orElse(retryableMessageSender);
+ Long min = Long.MAX_VALUE;
+ MessageSender sender = null;
+ for(Map.Entry<MessageSender, Long> entry :senders.entrySet()) {
+ if (entry.getValue() == Long.MAX_VALUE) {
+ continue;
+ } else {
+ if (min > entry.getValue()) {
+ min = entry.getValue();
+ sender = entry.getKey();
+ }
+ }
+ }
+ if (sender == null) {
+ return retryableMessageSender;
+ } else {
+ return sender;
+ }
}
private void scheduleReconnectTask(int reconnectDelay) {
- scheduler.scheduleWithFixedDelay(() -> {
- try {
- pendingTasks.take().run();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ scheduler.scheduleWithFixedDelay(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ pendingTasks.take().run();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
}
}, 0, reconnectDelay, MILLISECONDS);
}
- private Function<MessageSender, Runnable> errorHandlerFactory() {
- return messageSender -> {
- Runnable runnable = new PushBackReconnectRunnable(messageSender, senders, pendingTasks, availableMessageSenders);
- return () -> pendingTasks.offer(runnable);
- };
+ class ErrorHandlerFactory {
+ Runnable getHandler(MessageSender messageSender) {
+ final Runnable runnable = new PushBackReconnectRunnable(messageSender, senders, pendingTasks,
+ availableMessageSenders);
+ return new Runnable() {
+ @Override
+ public void run() {
+ pendingTasks.offer(runnable);
+ }
+ };
+ }
+
}
private static SslContext buildSslContext(AlphaClusterConfig clusterConfig) throws SSLException {
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
index abce82b..4eb90e1 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
@@ -34,6 +34,26 @@ public class RetryableMessageSender implements MessageSender {
}
@Override
+ public void onConnected() {
+
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
public AlphaResponse send(TxEvent event) {
if (event.type() == SagaStartedEvent) {
throw new OmegaException("Failed to process subsequent requests because no alpha server is available");
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
index b759052..822c7ec 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
@@ -28,6 +28,7 @@ import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Arrays;
+import java.util.concurrent.Callable;
import javax.net.ssl.SSLException;
@@ -65,7 +66,10 @@ public class LoadBalanceClusterMessageSenderWithTLSTest extends LoadBalancedClus
@BeforeClass
public static void beforeClass() throws Exception {
- Arrays.stream(ports).forEach(LoadBalanceClusterMessageSenderWithTLSTest::startServerOnPort);
+ for(int port: ports) {
+ System.out.println("Start the port " + port);
+ startServerOnPort(port);
+ }
}
private static void startServerOnPort(int port) {
@@ -122,7 +126,12 @@ public class LoadBalanceClusterMessageSenderWithTLSTest extends LoadBalancedClus
@Test
public void broadcastConnectionAndDisconnection() throws Exception {
messageSender.onConnected();
- await().atMost(1, SECONDS).until(() -> !connected.get(8080).isEmpty() && !connected.get(8090).isEmpty());
+ await().atMost(1, SECONDS).until( new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ return !connected.get(8080).isEmpty() && !connected.get(8090).isEmpty();
+ }
+ });
assertThat(connected.get(8080), contains("Connected " + serviceName));
assertThat(connected.get(8090), contains("Connected " + serviceName));
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 f26fb26..3a16558 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
@@ -37,6 +37,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
+import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.servicecomb.saga.common.EventType;
@@ -80,7 +81,9 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
@BeforeClass
public static void beforeClass() throws Exception {
- Arrays.stream(ports).forEach(LoadBalancedClusterMessageSenderTest::startServerOnPort);
+ for(int port: ports) {
+ startServerOnPort(port);
+ }
}
private static void startServerOnPort(int port) {
@@ -121,13 +124,19 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
messageSender.onConnected();
messageSender.send(event);
- int deadPort = killServerReceivedMessage();
+ final int deadPort = killServerReceivedMessage();
// ensure live message sender has latency greater than 0
messageSender.send(event);
startServerOnPort(deadPort);
- await().atMost(5, SECONDS).until(() -> connected.get(deadPort).size() == 3);
+ await().atMost(5, SECONDS).until(new Callable<Boolean>() {
+
+ @Override
+ public Boolean call() throws Exception {
+ return connected.get(deadPort).size() == 3;
+ }
+ });
TxEvent abortedEvent = new TxAbortedEvent(globalTxId, localTxId, parentTxId, compensationMethod, new RuntimeException("oops"));
messageSender.send(abortedEvent);
@@ -137,7 +146,13 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
assertThat(eventsMap.get(deadPort).poll().toString(), is(event.toString()));
assertThat(eventsMap.get(deadPort).poll().toString(), is(abortedEvent.toString()));
- await().atMost(3, SECONDS).until(() -> compensated.contains(globalTxId));
+ await().atMost(3, SECONDS).until(new Callable<Boolean>() {
+
+ @Override
+ public Boolean call() throws Exception {
+ return compensated.contains(globalTxId);
+ }
+ });
}
@Test(timeout = 1000)
@@ -145,9 +160,14 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
MessageSender underlying = Mockito.mock(MessageSender.class);
doThrow(RuntimeException.class).when(underlying).send(event);
- MessageSender messageSender = new LoadBalancedClusterMessageSender(underlying);
+ final MessageSender messageSender = new LoadBalancedClusterMessageSender(underlying);
- Thread thread = new Thread(() -> messageSender.send(event));
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ messageSender.send(event);
+ }
+ });
thread.start();
Thread.sleep(300);
@@ -162,7 +182,13 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
@Test
public void broadcastConnectionAndDisconnection() throws Exception {
messageSender.onConnected();
- await().atMost(1, SECONDS).until(() -> !connected.get(8080).isEmpty() && !connected.get(8090).isEmpty());
+ await().atMost(1, SECONDS).until(new Callable<Boolean>() {
+
+ @Override
+ public Boolean call() throws Exception {
+ return !connected.get(8080).isEmpty() && !connected.get(8090).isEmpty();
+ }
+ });
assertThat(connected.get(8080), contains("Connected " + serviceName));
assertThat(connected.get(8090), contains("Connected " + serviceName));
@@ -228,14 +254,26 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
@Test
public void stopSendingWhenClusterIsDown() throws Exception {
- servers.values().forEach(Server::shutdownNow);
+ for(Server server:servers.values()) {
+ server.shutdownNow();
+ }
messageSender.onConnected();
- Thread thread = new Thread(() -> messageSender.send(event));
+ final Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ messageSender.send(event);
+ }
+ });
thread.start();
// we don't want to keep sending on cluster down
- await().atMost(2, SECONDS).until(() -> thread.isAlive() && thread.getState().equals(WAITING));
+ await().atMost(2, SECONDS).until(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ return thread.isAlive() && thread.getState().equals(WAITING);
+ }
+ });
assertThat(eventsMap.get(8080).isEmpty(), is(true));
assertThat(eventsMap.get(8090).isEmpty(), is(true));
@@ -243,8 +281,18 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
startServerOnPort(8080);
startServerOnPort(8090);
- await().atMost(2, SECONDS).until(() -> connected.get(8080).size() == 2 || connected.get(8090).size() == 2);
- await().atMost(2, SECONDS).until(() -> eventsMap.get(8080).size() == 1 || eventsMap.get(8090).size() == 1);
+ await().atMost(2, SECONDS).until(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ return connected.get(8080).size() == 2 || connected.get(8090).size() == 2;
+ }
+ });
+ await().atMost(2, SECONDS).until(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ return eventsMap.get(8080).size() == 1 || eventsMap.get(8090).size() == 1;
+ }
+ });
}
@Test
@@ -257,12 +305,15 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
@Test
public void blowsUpWhenServerIsInterrupted() throws InterruptedException {
- Thread thread = new Thread(() -> {
- try {
- messageSender.send(event);
- expectFailing(OmegaException.class);
- } catch (OmegaException e) {
- assertThat(e.getMessage().endsWith("interruption"), is(true));
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ messageSender.send(event);
+ expectFailing(OmegaException.class);
+ } catch (OmegaException e) {
+ assertThat(e.getMessage().endsWith("interruption"), is(true));
+ }
}
});
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
index 00c8bef..9a2db23 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
@@ -57,13 +57,13 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
}};
protected static final Map<Integer, Queue<String>> connected = new HashMap<Integer, Queue<String>>() {{
- put(8080, new ConcurrentLinkedQueue<>());
- put(8090, new ConcurrentLinkedQueue<>());
+ put(8080, new ConcurrentLinkedQueue<String>());
+ put(8090, new ConcurrentLinkedQueue<String>());
}};
protected static final Map<Integer, Queue<TxEvent>> eventsMap = new HashMap<Integer, Queue<TxEvent>>() {{
- put(8080, new ConcurrentLinkedQueue<>());
- put(8090, new ConcurrentLinkedQueue<>());
+ put(8080, new ConcurrentLinkedQueue<TxEvent>());
+ put(8090, new ConcurrentLinkedQueue<TxEvent>());
}};
protected final List<String> compensated = new ArrayList<>();
@@ -81,12 +81,30 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
protected final String serviceName = uniquify("serviceName");
- protected final MessageSerializer serializer = objects -> objects[0].toString().getBytes();
+ protected final MessageSerializer serializer = new MessageSerializer() {
+ @Override
+ public byte[] serialize(Object[] objects) {
+ return objects[0].toString().getBytes();
+ }
+ };
+
+ protected final MessageDeserializer deserializer = new MessageDeserializer() {
+
+ @Override
+ public Object[] deserialize(byte[] message) {
+ return new Object[] {new String(message)};
+ }
+ };
- protected final MessageDeserializer deserializer = message -> new Object[] {new String(message)};
+ protected final MessageHandler handler = new MessageHandler() {
- protected final MessageHandler handler = (globalTxId, localTxId, parentTxId, compensationMethod,
- payloads) -> compensated.add(globalTxId);
+ @Override
+ public void onReceive(String globalTxId, String localTxId, String parentTxId, String compensationMethod,
+ Object... payloads) {
+ compensated.add(globalTxId);
+
+ }
+ };
protected final String[] addresses = {"localhost:8080", "localhost:8090"};
@@ -94,7 +112,9 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
@AfterClass
public static void tearDown() throws Exception {
- servers.values().forEach(Server::shutdown);
+ for(Server server: servers.values()) {
+ server.shutdown();
+ }
}
protected abstract MessageSender newMessageSender(String[] addresses);
@@ -103,8 +123,12 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
public void after() throws Exception {
messageSender.onDisconnected();
messageSender.close();
- eventsMap.values().forEach(Queue::clear);
- connected.values().forEach(Queue::clear);
+ for (Queue<TxEvent> queue :eventsMap.values()) {
+ queue.clear();
+ }
+ for (Queue<String> queue :connected.values()) {
+ queue.clear();
+ }
}
protected static class MyTxEventService extends TxEventServiceImplBase {
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
index e658946..4a12f1f 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
@@ -31,11 +31,16 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.hamcrest.Matcher;
import org.junit.Before;
import org.junit.Test;
public class PushBackReconnectRunnableTest {
- private static final Runnable NO_OP_RUNNABLE = () -> {
+ private static final Runnable NO_OP_RUNNABLE = new Runnable() {
+ @Override
+ public void run() {
+ // Do nothing here
+ }
};
private final MessageSender sender = mock(MessageSender.class);
@@ -60,11 +65,11 @@ public class PushBackReconnectRunnableTest {
assertThat(runnables, contains(NO_OP_RUNNABLE, pushBack));
assertThat(runnables.poll(), is(NO_OP_RUNNABLE));
- assertThat(runnables, contains(pushBack));
+ assertThat(runnables.contains(pushBack), is(true));
// failed again and pushed back itself to queue
runnables.poll().run();
- assertThat(runnables, contains(pushBack));
+ assertThat(runnables.contains(pushBack), is(true));
runnables.poll().run();
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
index 3856415..c1edb26 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
@@ -70,12 +70,15 @@ public class RetryableMessageSenderTest {
@Test
public void blowsUpWhenInterrupted() throws InterruptedException {
- Thread thread = new Thread(() -> {
- try {
- messageSender.send(event);
- expectFailing(OmegaException.class);
- } catch (OmegaException e) {
- assertThat(e.getMessage().endsWith("interruption"), is(true));
+ Thread thread = new Thread( new Runnable() {
+ @Override
+ public void run() {
+ try {
+ messageSender.send(event);
+ expectFailing(OmegaException.class);
+ } catch (OmegaException e) {
+ assertThat(e.getMessage().endsWith("interruption"), is(true));
+ }
}
});
@@ -83,4 +86,4 @@ public class RetryableMessageSenderTest {
thread.interrupt();
thread.join();
}
-}
\ No newline at end of file
+}
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
index cc43865..75d5864 100644
--- a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
@@ -36,6 +36,7 @@ import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -48,6 +49,7 @@ import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxAbortedEvent;
import org.apache.servicecomb.saga.omega.transaction.TxCompensatedEvent;
import org.apache.servicecomb.saga.omega.transaction.TxEndedEvent;
+import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.omega.transaction.TxStartedEvent;
import org.apache.servicecomb.saga.omega.transaction.spring.TransactionInterceptionTest.MessageConfig;
import org.apache.servicecomb.saga.omega.transaction.spring.annotations.OmegaContextAware;
@@ -65,11 +67,14 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.junit4.SpringRunner;
import io.reactivex.Flowable;
+import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import akka.actor.AbstractLoggingActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
+import akka.japi.Creator;
+import akka.japi.pf.FI.UnitApply;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {TransactionTestMain.class, MessageConfig.class})
@@ -222,7 +227,10 @@ public class TransactionInterceptionTest {
is(new TxEndedEvent(globalTxId, newLocalTxId, globalTxId, compensationMethod2).toString()));
assertThat(userRepository.count(), is(1L));
- userRepository.findAll().forEach(user -> assertThat(user, is(this.user)));
+ Iterable<User> users = userRepository.findAll();
+ for(User user: users ) {
+ assertThat(user, is(this.user));
+ }
}
@Test
@@ -257,7 +265,7 @@ public class TransactionInterceptionTest {
new Thread(new Runnable() {
@Override
public void run() {
- userService.add(user))
+ userService.add(user);
}
}).start();
waitTillSavedUser(username);
@@ -283,11 +291,21 @@ public class TransactionInterceptionTest {
@Test
public void passesOmegaContextInThreadPool() throws Exception {
- executor.schedule(() -> userService.add(user), 0, MILLISECONDS);
+ executor.schedule(new Runnable() {
+ @Override
+ public void run() {
+ userService.add(user);
+ }
+ }, 0, MILLISECONDS);
waitTillSavedUser(username);
String localTxId = omegaContext.newLocalTxId();
- executor.invokeAll(singletonList(() -> userService.add(jack)));
+ executor.invokeAll(singletonList(new Callable<User>() {
+ @Override
+ public User call() throws Exception {
+ return userService.add(jack);
+ }
+ }));
waitTillSavedUser(usernameJack);
assertArrayEquals(
@@ -306,7 +324,12 @@ public class TransactionInterceptionTest {
Flowable.just(user)
.parallel()
.runOn(Schedulers.io())
- .doOnNext(userService::add)
+ .doOnNext(new Consumer() {
+ @Override
+ public void accept(Object user) throws Exception {
+ userService.add((User)user);
+ }
+ })
.sequential()
.subscribe();
@@ -339,8 +362,13 @@ public class TransactionInterceptionTest {
actorSystem.terminate();
}
- private void waitTillSavedUser(String username) {
- await().atMost(1000, MILLISECONDS).until(() -> userRepository.findByUsername(username) != null);
+ private void waitTillSavedUser(final String username) {
+ await().atMost(1000, MILLISECONDS).until(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ return userRepository.findByUsername(username) != null;
+ }
+ });
}
private static class UserServiceActor extends AbstractLoggingActor {
@@ -350,15 +378,24 @@ public class TransactionInterceptionTest {
this.userService = userService;
}
- static Props props(TransactionalUserService userService) {
- return Props.create(UserServiceActor.class, () -> new UserServiceActor(userService));
+ static Props props(final TransactionalUserService userService) {
+ return Props.create(UserServiceActor.class, new Creator<UserServiceActor>() {
+ @Override
+ public UserServiceActor create() throws Exception {
+ return new UserServiceActor(userService);
+ }
+ });
}
@Override
public Receive createReceive() {
return receiveBuilder()
- .match(User.class, userService::add)
- .build();
+ .match(User.class, new UnitApply<User>() {
+ @Override
+ public void apply(User user) throws Exception {
+ userService.add(user);
+ }
+ }).build();
}
}
@@ -387,9 +424,32 @@ public class TransactionInterceptionTest {
@Bean
MessageSender sender() {
- return (event) -> {
- messages.add(event.toString());
- return new AlphaResponse(false);
+ return new MessageSender() {
+ @Override
+ public void onConnected() {
+
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOW";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event.toString());
+ return new AlphaResponse(false);
+ }
};
}
}
[incubator-servicecomb-saga] 01/06: Updated the gitter room address
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 316af594f64023da5feb732b4537924b2ad8950c
Author: Willem Jiang <ji...@huawei.com>
AuthorDate: Mon Jul 30 08:50:00 2018 +0800
Updated the gitter room address
---
README.md | 2 +-
README_ZH.md | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/README.md b/README.md
index 68ff146..45ee3ef 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,4 @@
-# Saga | [中文](README_ZH.md) [](https://travis-ci.org/apache/incubator-servicecomb-saga?branch=master) [](https://coveralls.io/github/apache/incubator-servicecomb-saga?branch=master)[ [...]
+# Saga | [中文](README_ZH.md) [](https://travis-ci.org/apache/incubator-servicecomb-saga?branch=master) [](https://coveralls.io/github/apache/incubator-servicecomb-saga?branch=master)[ [...]
Apache ServiceComb (incubating) Saga is an eventually data consistency solution for micro-service applications. Transactions are commited directly in the try phase and compensated in reverse order in the rollback phase comparing to [TCC](http://design.inf.usi.ch/sites/default/files/biblio/rest-tcc.pdf).
## Features
diff --git a/README_ZH.md b/README_ZH.md
index 86b9f97..5bfc709 100644
--- a/README_ZH.md
+++ b/README_ZH.md
@@ -1,4 +1,4 @@
-# Saga | [English](README.md) [](https://travis-ci.org/apache/incubator-servicecomb-saga?branch=master) [](https://coveralls.io/github/apache/incubator-servicecomb-saga?branch=master) [](https://www.apache.org/licenses/LI [...]
+# Saga | [English](README.md) [](https://travis-ci.org/apache/incubator-servicecomb-saga?branch=master) [](https://coveralls.io/github/apache/incubator-servicecomb-saga?branch=master) [](https://www.apache.org/licenses/LI [...]
Apache ServiceComb (incubating) Saga 是一个微服务应用的数据最终一致性解决方案。相对于[TCC](http://design.inf.usi.ch/sites/default/files/biblio/rest-tcc.pdf)而言，在try阶段，Saga会直接提交事务，后续rollback阶段则通过反向的补偿操作来完成。
## 特性
[incubator-servicecomb-saga] 03/06: SCB-648 Update the
omega-transport modules
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 f20f1442af4e79ac8f3dd0bb2568e147d3355cd4
Author: Willem Jiang <ji...@huawei.com>
AuthorDate: Tue Jul 31 15:15:45 2018 +0800
SCB-648 Update the omega-transport modules
---
.../src/test/java/SagaDubboConsumerFilterTest.java | 9 ++++++++-
.../src/test/java/SagaDubboProviderFilterTest.java | 8 +++++++-
.../transport/feign/FeignAutoConfigurationTest.java | 19 +++++++++++++++----
.../TransactionHandlerInterceptorTest.java | 9 ++++++++-
.../servicecomb/SagaConsumerHandlerTest.java | 8 +++++++-
.../servicecomb/SagaProviderHandlerTest.java | 12 ++++++++++--
6 files changed, 55 insertions(+), 10 deletions(-)
diff --git a/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboConsumerFilterTest.java b/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboConsumerFilterTest.java
index 3fcb1b1..dc1875b 100644
--- a/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboConsumerFilterTest.java
+++ b/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboConsumerFilterTest.java
@@ -19,6 +19,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
+import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transport.dubbo.SagaDubboConsumerFilter;
import org.junit.Before;
@@ -36,7 +37,13 @@ public class SagaDubboConsumerFilterTest {
private static final String globalTxId = UUID.randomUUID().toString();
private static final String localTxId = UUID.randomUUID().toString();
- private final OmegaContext omegaContext = new OmegaContext(() -> "ignored");
+ private final OmegaContext omegaContext = new OmegaContext(new IdGenerator<String>() {
+
+ @Override
+ public String nextId() {
+ return "ignored";
+ }
+ });
private final Invocation invocation = mock(Invocation.class);
private final SagaDubboConsumerFilter filter = new SagaDubboConsumerFilter();
diff --git a/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboProviderFilterTest.java b/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboProviderFilterTest.java
index be00e69..1d0e8ad 100644
--- a/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboProviderFilterTest.java
+++ b/omega/omega-transport/omega-transport-dubbo/src/test/java/SagaDubboProviderFilterTest.java
@@ -17,6 +17,7 @@
import java.util.UUID;
+import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transport.dubbo.SagaDubboProviderFilter;
import org.junit.Before;
@@ -34,7 +35,12 @@ public class SagaDubboProviderFilterTest {
private static final String globalTxId = UUID.randomUUID().toString();
private static final String localTxId = UUID.randomUUID().toString();
- private final OmegaContext omegaContext = new OmegaContext(() -> "ignored");
+ private final OmegaContext omegaContext = new OmegaContext(new IdGenerator<String>() {
+ @Override
+ public String nextId() {
+ return "ignored";
+ }
+ });
private final Invocation invocation = mock(Invocation.class);
private final SagaDubboProviderFilter filter = new SagaDubboProviderFilter();
diff --git a/omega/omega-transport/omega-transport-feign/src/test/java/org/apache/servicecomb/saga/omega/transport/feign/FeignAutoConfigurationTest.java b/omega/omega-transport/omega-transport-feign/src/test/java/org/apache/servicecomb/saga/omega/transport/feign/FeignAutoConfigurationTest.java
index f10f89a..4fa35ac 100644
--- a/omega/omega-transport/omega-transport-feign/src/test/java/org/apache/servicecomb/saga/omega/transport/feign/FeignAutoConfigurationTest.java
+++ b/omega/omega-transport/omega-transport-feign/src/test/java/org/apache/servicecomb/saga/omega/transport/feign/FeignAutoConfigurationTest.java
@@ -27,7 +27,10 @@ import java.util.UUID;
import feign.RequestInterceptor;
import feign.RequestTemplate;
+
+import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
+import org.hamcrest.Matchers;
import org.junit.Before;
import org.junit.Test;
@@ -37,7 +40,13 @@ public class FeignAutoConfigurationTest{
private static final String localTxId = UUID.randomUUID().toString();
- private final OmegaContext omegaContext = new OmegaContext(() -> "ignored");
+ private final OmegaContext omegaContext = new OmegaContext(new IdGenerator<String>() {
+
+ @Override
+ public String nextId() {
+ return "ignored";
+ }
+ });
private RequestTemplate requestTemplate = new RequestTemplate(); // mock(RequestTemplate.class);
@@ -55,8 +64,10 @@ public class FeignAutoConfigurationTest{
feignClientRequestInterceptor.apply(requestTemplate); // .preHandle(request, response, null);
- assertThat((new ArrayList(requestTemplate.headers().get(OmegaContext.GLOBAL_TX_ID_KEY))).get(0), is(globalTxId));
- assertThat((new ArrayList(requestTemplate.headers().get(OmegaContext.LOCAL_TX_ID_KEY))).get(0), is(localTxId));
+ assertThat((new ArrayList(requestTemplate.headers().get(OmegaContext.GLOBAL_TX_ID_KEY))).get(0),
+ Matchers.<Object>is(globalTxId));
+ assertThat((new ArrayList(requestTemplate.headers().get(OmegaContext.LOCAL_TX_ID_KEY))).get(0),
+ Matchers.<Object>is(localTxId));
}
@Test
@@ -69,4 +80,4 @@ public class FeignAutoConfigurationTest{
assertThat(requestTemplate.headers().get(OmegaContext.GLOBAL_TX_ID_KEY), is(nullValue()));
assertThat(requestTemplate.headers().get(OmegaContext.LOCAL_TX_ID_KEY), is(nullValue()));
}
-}
\ No newline at end of file
+}
diff --git a/omega/omega-transport/omega-transport-resttemplate/src/test/java/org/apache/servicecomb/saga/omega/transport/resttemplate/TransactionHandlerInterceptorTest.java b/omega/omega-transport/omega-transport-resttemplate/src/test/java/org/apache/servicecomb/saga/omega/transport/resttemplate/TransactionHandlerInterceptorTest.java
index 7b41540..16cbe6d 100644
--- a/omega/omega-transport/omega-transport-resttemplate/src/test/java/org/apache/servicecomb/saga/omega/transport/resttemplate/TransactionHandlerInterceptorTest.java
+++ b/omega/omega-transport/omega-transport-resttemplate/src/test/java/org/apache/servicecomb/saga/omega/transport/resttemplate/TransactionHandlerInterceptorTest.java
@@ -30,6 +30,7 @@ import java.util.UUID;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
+import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.junit.Before;
import org.junit.Test;
@@ -41,7 +42,13 @@ public class TransactionHandlerInterceptorTest {
private static final String localTxId = UUID.randomUUID().toString();
- private final OmegaContext omegaContext = new OmegaContext(() -> "ignored");
+ private final OmegaContext omegaContext = new OmegaContext(new IdGenerator<String>() {
+
+ @Override
+ public String nextId() {
+ return "ignored";
+ }
+ });
private HandlerInterceptor requestInterceptor = new TransactionHandlerInterceptor(omegaContext);
diff --git a/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaConsumerHandlerTest.java b/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaConsumerHandlerTest.java
index d66e3ee..18b57ac 100644
--- a/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaConsumerHandlerTest.java
+++ b/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaConsumerHandlerTest.java
@@ -42,7 +42,13 @@ public class SagaConsumerHandlerTest {
@SuppressWarnings("unchecked")
private final IdGenerator<String> idGenerator = mock(IdGenerator.class);
- private final OmegaContext omegaContext = new OmegaContext(() -> "ignored");
+ private final OmegaContext omegaContext = new OmegaContext(new IdGenerator<String>() {
+
+ @Override
+ public String nextId() {
+ return "ignored";
+ }
+ });
private final Invocation invocation = mock(Invocation.class);
diff --git a/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaProviderHandlerTest.java b/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaProviderHandlerTest.java
index 24c277b..41fc049 100644
--- a/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaProviderHandlerTest.java
+++ b/omega/omega-transport/omega-transport-servicecomb/src/test/java/org/apache/servicecomb/saga/omega/transport/servicecomb/SagaProviderHandlerTest.java
@@ -26,11 +26,13 @@ import static org.junit.Assert.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.apache.servicecomb.core.Invocation;
+import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.swagger.invocation.AsyncResponse;
import org.junit.Before;
@@ -42,7 +44,13 @@ public class SagaProviderHandlerTest {
private static final String localTxId = UUID.randomUUID().toString();
- private final OmegaContext omegaContext = new OmegaContext(() -> "ignored");
+ private final OmegaContext omegaContext = new OmegaContext(new IdGenerator<String>() {
+
+ @Override
+ public String nextId() {
+ return "ignored";
+ }
+ });
private final Invocation invocation = mock(Invocation.class);
@@ -70,7 +78,7 @@ public class SagaProviderHandlerTest {
@Test
public void doNothingInNonTransactionRequest() throws Exception {
- when(invocation.getContext()).thenReturn(emptyMap());
+ when(invocation.getContext()).thenReturn(Collections.EMPTY_MAP);
handler.handle(invocation, asyncResponse);