You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@servicecomb.apache.org by GitBox <gi...@apache.org> on 2018/01/16 00:56:00 UTC
[GitHub] seanyinx closed pull request #114: SCB-227 stop sub tx from running when global tx failed
seanyinx closed pull request #114: SCB-227 stop sub tx from running when global tx failed
URL: https://github.com/apache/incubator-servicecomb-saga/pull/114
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/TxConsistentService.java b/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/TxConsistentService.java
index bddba36d..fa937524 100644
--- a/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/TxConsistentService.java
+++ b/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/TxConsistentService.java
@@ -54,7 +54,11 @@ public TxConsistentService(TxEventRepository eventRepository, OmegaCallback omeg
this.omegaCallback = omegaCallback;
}
- public void handle(TxEvent event) {
+ public boolean handle(TxEvent event) {
+ if (TxStartedEvent.name().equals(event.type()) && isGlobalTxAborted(event)) {
+ return false;
+ }
+
eventRepository.save(event);
executor.execute(() -> {
@@ -64,6 +68,8 @@ public void handle(TxEvent event) {
eventCallbacks.getOrDefault(event.type(), DO_NOTHING_CONSUMER).accept(event);
});
+
+ return true;
}
private void compensate(TxEvent event) {
diff --git a/alpha/alpha-core/src/test/java/org/apache/servicecomb/saga/alpha/core/TxConsistentServiceTest.java b/alpha/alpha-core/src/test/java/org/apache/servicecomb/saga/alpha/core/TxConsistentServiceTest.java
index a44f16da..9318a069 100644
--- a/alpha/alpha-core/src/test/java/org/apache/servicecomb/saga/alpha/core/TxConsistentServiceTest.java
+++ b/alpha/alpha-core/src/test/java/org/apache/servicecomb/saga/alpha/core/TxConsistentServiceTest.java
@@ -135,6 +135,19 @@ public void compensateTxEndedEventImmediately_IfGlobalTxAlreadyFailed() throws E
));
}
+ @Test
+ public void skipTxStartedEvent_IfGlobalTxAlreadyFailed() {
+ String localTxId1 = UUID.randomUUID().toString();
+ events.add(newEvent(TxStartedEvent));
+ events.add(newEvent(TxAbortedEvent));
+
+ TxEvent event = eventOf(TxStartedEvent, "service x".getBytes(), localTxId1, "method x");
+
+ consistentService.handle(event);
+
+ assertThat(events.size(), is(2));
+ }
+
private TxEvent newEvent(EventType eventType) {
return new TxEvent(serviceName, instanceId, new Date(), globalTxId, localTxId, parentTxId, eventType.name(), compensationMethod, "yeah".getBytes());
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java
index 4b3ea1b7..eced7f9f 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java
@@ -39,7 +39,9 @@
class GrpcTxEventEndpointImpl extends TxEventServiceImplBase {
- private static final GrpcAck ACK = GrpcAck.newBuilder().build();
+ private static final GrpcAck ALLOW = GrpcAck.newBuilder().setAborted(false).build();
+ private static final GrpcAck REJECT = GrpcAck.newBuilder().setAborted(true).build();
+
private final TxConsistentService txConsistentService;
private final Map<String, Map<String, OmegaCallback>> omegaCallbacks;
@@ -67,13 +69,13 @@ public void onDisconnected(GrpcServiceConfig request, StreamObserver<GrpcAck> re
callback.disconnect();
}
- responseObserver.onNext(ACK);
+ responseObserver.onNext(ALLOW);
responseObserver.onCompleted();
}
@Override
public void onTxEvent(GrpcTxEvent message, StreamObserver<GrpcAck> responseObserver) {
- txConsistentService.handle(new TxEvent(
+ boolean ok = txConsistentService.handle(new TxEvent(
message.getServiceName(),
message.getInstanceId(),
new Date(message.getTimestamp()),
@@ -85,7 +87,7 @@ public void onTxEvent(GrpcTxEvent message, StreamObserver<GrpcAck> responseObser
message.getPayloads().toByteArray()
));
- responseObserver.onNext(ACK);
+ responseObserver.onNext(ok ? ALLOW : REJECT);
responseObserver.onCompleted();
}
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
index a501f759..6cf18938 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
@@ -38,6 +38,7 @@
import org.apache.servicecomb.saga.alpha.core.OmegaCallback;
import org.apache.servicecomb.saga.alpha.core.TxConsistentService;
import org.apache.servicecomb.saga.alpha.core.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;
@@ -250,6 +251,24 @@ public void compensateOnlyFailedGlobalTransaction() {
anotherBlockingStub.onDisconnected(anotherServiceConfig);
}
+ @Test
+ public void doNotStartSubTxOnFailure() {
+ asyncStub.onConnected(serviceConfig, compensateResponseObserver);
+
+ blockingStub.onTxEvent(eventOf(TxStartedEvent, localTxId, parentTxId, "service a".getBytes(), "method a"));
+
+ blockingStub.onTxEvent(someGrpcEvent(TxAbortedEvent));
+
+ await().atMost(1, SECONDS).until(() -> receivedCommands.size() == 1);
+
+ String localTxId1 = UUID.randomUUID().toString();
+ String parentTxId1 = UUID.randomUUID().toString();
+ GrpcAck result = blockingStub
+ .onTxEvent(eventOf(TxStartedEvent, localTxId1, parentTxId1, "service b".getBytes(), "method b"));
+
+ assertThat(result.getAborted(), is(true));
+ }
+
private GrpcServiceConfig someServiceConfig() {
return GrpcServiceConfig.newBuilder()
.setServiceName(uniquify("serviceName"))
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 f6b5d763..4af0773f 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
@@ -23,11 +23,13 @@
import java.util.function.Function;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
+import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
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.GrpcServiceConfig;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTxEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTxEvent.Builder;
@@ -83,8 +85,9 @@ public String target() {
}
@Override
- public void send(TxEvent event) {
- blockingEventService.onTxEvent(convertEvent(event));
+ public AlphaResponse send(TxEvent event) {
+ GrpcAck grpcAck = blockingEventService.onTxEvent(convertEvent(event));
+ return new AlphaResponse(grpcAck.getAborted());
}
private GrpcTxEvent convertEvent(TxEvent 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 73940ed7..700864af 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
@@ -34,10 +34,12 @@
import java.util.function.Function;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
+import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
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.OmegaException;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -124,24 +126,27 @@ public void close() {
}
@Override
- public void send(TxEvent event) {
- boolean success = false;
+ public AlphaResponse send(TxEvent event) {
do {
MessageSender messageSender = fastestSender();
try {
long startTime = System.nanoTime();
- messageSender.send(event);
+ AlphaResponse response = messageSender.send(event);
senders.put(messageSender, System.nanoTime() - startTime);
- success = true;
+ return response;
+ } catch (OmegaException e) {
+ throw e;
} catch (Exception e) {
log.error("Retry sending event {} due to failure", event, e);
// very large latency on exception
senders.put(messageSender, Long.MAX_VALUE);
}
- } while (!success && !Thread.currentThread().isInterrupted());
+ } while (!Thread.currentThread().isInterrupted());
+
+ throw new OmegaException("Failed to send event " + event + " due to interruption");
}
private MessageSender fastestSender() {
@@ -150,13 +155,13 @@ private MessageSender fastestSender() {
.filter(entry -> entry.getValue() < Long.MAX_VALUE)
.min(Comparator.comparingLong(Entry::getValue))
.map(Entry::getKey)
- .orElse((event -> {
+ .orElse(event -> {
try {
- availableMessageSenders.take().send(event);
+ return availableMessageSenders.take().send(event);
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ throw new OmegaException("Failed to send event " + event + " due to interruption", e);
}
- }));
+ });
}
private void scheduleReconnectTask(int reconnectDelay) {
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 1fdfd2e7..93cb8544 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
@@ -45,8 +45,10 @@
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.OmegaException;
import org.apache.servicecomb.saga.omega.transaction.TxAbortedEvent;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
+import org.apache.servicecomb.saga.omega.transaction.TxStartedEvent;
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;
@@ -293,6 +295,30 @@ public void stopSendingWhenClusterIsDown() throws Exception {
await().atMost(2, SECONDS).until(() -> eventsMap.get(8080).size() == 1 || eventsMap.get(8090).size() == 1);
}
+ @Test
+ public void forwardSendResult() {
+ assertThat(messageSender.send(event).aborted(), is(false));
+
+ TxEvent rejectEvent = new TxStartedEvent(globalTxId, localTxId, parentTxId, "reject", "blah");
+ assertThat(messageSender.send(rejectEvent).aborted(), is(true));
+ }
+
+ @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.start();
+ thread.interrupt();
+ thread.join();
+ }
+
private int killServerReceivedMessage() {
for (int port : eventsMap.keySet()) {
if (!eventsMap.get(port).isEmpty()) {
@@ -341,7 +367,12 @@ public void onTxEvent(GrpcTxEvent request, StreamObserver<GrpcAck> responseObser
.build());
}
- responseObserver.onNext(GrpcAck.newBuilder().build());
+ if ("TxStartedEvent".equals(request.getType()) && request.getCompensationMethod().equals("reject")) {
+ responseObserver.onNext(GrpcAck.newBuilder().setAborted(true).build());
+ } else {
+ responseObserver.onNext(GrpcAck.newBuilder().setAborted(false).build());
+ }
+
responseObserver.onCompleted();
}
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 c30953f8..7daf9547 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
@@ -39,6 +39,7 @@
import org.apache.servicecomb.saga.omega.context.CompensationContext;
import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
+import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxAbortedEvent;
@@ -310,7 +311,10 @@ OmegaContext omegaContext() {
@Bean
MessageSender sender() {
- return (event) -> messages.add(event.toString());
+ return (event) -> {
+ messages.add(event.toString());
+ return new AlphaResponse(false);
+ };
}
}
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/AlphaResponse.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/AlphaResponse.java
new file mode 100644
index 00000000..07df7937
--- /dev/null
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/AlphaResponse.java
@@ -0,0 +1,30 @@
+/*
+ * 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.transaction;
+
+public class AlphaResponse {
+ private final boolean aborted;
+
+ public AlphaResponse(boolean aborted) {
+ this.aborted = aborted;
+ }
+
+ public boolean aborted() {
+ return aborted;
+ }
+}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java
index 76193cd2..074a5eca 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java
@@ -29,8 +29,9 @@
}
@Override
- public void preIntercept(String parentTxId, String compensationMethod, Object... message) {
- sender.send(new TxStartedEvent(context.globalTxId(), context.localTxId(), parentTxId, compensationMethod, message));
+ public AlphaResponse preIntercept(String parentTxId, String compensationMethod, Object... message) {
+ return sender
+ .send(new TxStartedEvent(context.globalTxId(), context.localTxId(), parentTxId, compensationMethod, message));
}
@Override
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/EventAwareInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/EventAwareInterceptor.java
index 291538fc..5f8165f5 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/EventAwareInterceptor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/EventAwareInterceptor.java
@@ -20,7 +20,8 @@
public interface EventAwareInterceptor {
EventAwareInterceptor NO_OP_INTERCEPTOR = new EventAwareInterceptor() {
@Override
- public void preIntercept(String parentTxId, String compensationMethod, Object... message) {
+ public AlphaResponse preIntercept(String parentTxId, String compensationMethod, Object... message) {
+ return new AlphaResponse(false);
}
@Override
@@ -32,7 +33,7 @@ public void onError(String parentTxId, String compensationMethod, Throwable thro
}
};
- void preIntercept(String parentTxId, String compensationMethod, Object... message);
+ AlphaResponse preIntercept(String parentTxId, String compensationMethod, Object... message);
void postIntercept(String parentTxId, String compensationMethod);
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 2f053942..5271b823 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,8 +18,7 @@
package org.apache.servicecomb.saga.omega.transaction;
public interface MessageSender {
- MessageSender NO_OP_SENDER = event -> {
- };
+ MessageSender NO_OP_SENDER = event -> new AlphaResponse(false);
default void onConnected() {
}
@@ -34,5 +33,5 @@ default String target() {
return "UNKNOWN";
}
- void send(TxEvent event);
+ AlphaResponse send(TxEvent event);
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/OmegaException.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/OmegaException.java
index 89eb67f4..4488e447 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/OmegaException.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/OmegaException.java
@@ -18,6 +18,9 @@
package org.apache.servicecomb.saga.omega.transaction;
public class OmegaException extends RuntimeException {
+ public OmegaException(String message) {
+ super(message);
+ }
public OmegaException(String cause, Throwable throwable) {
super(cause, throwable);
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java
index 2dad5ae3..7299b255 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java
@@ -30,8 +30,8 @@
}
@Override
- public void preIntercept(String parentTxId, String compensationMethod, Object... message) {
- sender.send(new SagaStartedEvent(omegaContext.globalTxId(), omegaContext.localTxId()));
+ public AlphaResponse preIntercept(String parentTxId, String compensationMethod, Object... message) {
+ return sender.send(new SagaStartedEvent(omegaContext.globalTxId(), omegaContext.localTxId()));
}
@Override
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TimeAwareInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TimeAwareInterceptor.java
index 80ad03ff..ca145512 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TimeAwareInterceptor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TimeAwareInterceptor.java
@@ -29,8 +29,8 @@
}
@Override
- public void preIntercept(String parentTxId, String signature, Object... args) {
- interceptor.preIntercept(parentTxId, signature, args);
+ public AlphaResponse preIntercept(String parentTxId, String signature, Object... args) {
+ return interceptor.preIntercept(parentTxId, signature, args);
}
@Override
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java
index d3e091c9..cead07a5 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java
@@ -24,6 +24,8 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import javax.transaction.InvalidTransactionException;
+
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transaction.annotations.Compensable;
import org.aspectj.lang.ProceedingJoinPoint;
@@ -57,7 +59,13 @@ Object advise(ProceedingJoinPoint joinPoint, Compensable compensable) throws Thr
context.newLocalTxId();
TimeAwareInterceptor interceptor = new TimeAwareInterceptor(this.interceptor);
- interceptor.preIntercept(localTxId, signature, joinPoint.getArgs());
+ AlphaResponse response = interceptor.preIntercept(localTxId, signature, joinPoint.getArgs());
+ if (response.aborted()) {
+ String abortedLocalTxId = context.localTxId();
+ context.setLocalTxId(localTxId);
+ throw new InvalidTransactionException("Abort sub transaction " + abortedLocalTxId +
+ " because global transaction " + context.globalTxId() + " has already aborted.");
+ }
LOG.debug("Updated context {} for compensable method {} ", context, method.toString());
scheduleTimeoutTask(interceptor, localTxId, signature, method, compensable.timeout());
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 d4bfef67..21af7e6e 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
@@ -41,7 +41,10 @@
private final String localTxId = UUID.randomUUID().toString();
private final String parentTxId = UUID.randomUUID().toString();
- private final MessageSender sender = messages::add;
+ private final MessageSender sender = e -> {
+ messages.add(e);
+ return new AlphaResponse(false);
+ };
private final String message = uniquify("message");
private final String compensationMethod = getClass().getCanonicalName();
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 bc126ae7..0b33d4b8 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
@@ -33,7 +33,10 @@
public class CompensationMessageHandlerTest {
private final List<TxEvent> events = new ArrayList<>();
- private final MessageSender sender = events::add;
+ private final MessageSender sender = e -> {
+ events.add(e);
+ return new AlphaResponse(false);
+ };
private final String globalTxId = uniquify("globalTxId");
private final String localTxId = uniquify("localTxId");
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 c146f5bf..f8e936da 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
@@ -36,7 +36,10 @@
private final List<TxEvent> messages = new ArrayList<>();
- private final MessageSender sender = messages::add;
+ private final MessageSender sender = e -> {
+ messages.add(e);
+ return new AlphaResponse(false);
+ };
private final String globalTxId = UUID.randomUUID().toString();
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 6d2e1a6d..7316161c 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
@@ -47,7 +47,10 @@
private final List<TxEvent> messages = new ArrayList<>();
private final String globalTxId = UUID.randomUUID().toString();
- private final MessageSender sender = messages::add;
+ private final MessageSender sender = e -> {
+ messages.add(e);
+ 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/TimeAwareInterceptorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TimeAwareInterceptorTest.java
index d26f04f5..9ff0214c 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TimeAwareInterceptorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TimeAwareInterceptorTest.java
@@ -45,7 +45,8 @@
private final EventAwareInterceptor underlying = new EventAwareInterceptor() {
@Override
- public void preIntercept(String parentTxId, String compensationMethod, Object... message) {
+ public AlphaResponse preIntercept(String parentTxId, String compensationMethod, Object... message) {
+ return new AlphaResponse(false);
}
@Override
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 63cdff06..a3db9409 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
@@ -23,6 +23,10 @@
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
@@ -32,6 +36,8 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import javax.transaction.InvalidTransactionException;
+
import org.apache.servicecomb.saga.common.EventType;
import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
@@ -40,7 +46,6 @@
import org.aspectj.lang.reflect.MethodSignature;
import org.junit.Before;
import org.junit.Test;
-import org.mockito.Mockito;
public class TransactionAspectTest {
private final List<TxEvent> messages = new ArrayList<>();
@@ -49,13 +54,16 @@
private final String newLocalTxId = UUID.randomUUID().toString();
- private final MessageSender sender = messages::add;
- private final ProceedingJoinPoint joinPoint = Mockito.mock(ProceedingJoinPoint.class);
- private final MethodSignature methodSignature = Mockito.mock(MethodSignature.class);
+ private final MessageSender sender = e -> {
+ messages.add(e);
+ return new AlphaResponse(false);
+ };
+ private final ProceedingJoinPoint joinPoint = mock(ProceedingJoinPoint.class);
+ private final MethodSignature methodSignature = mock(MethodSignature.class);
@SuppressWarnings("unchecked")
- private final IdGenerator<String> idGenerator = Mockito.mock(IdGenerator.class);
- private final Compensable compensable = Mockito.mock(Compensable.class);
+ private final IdGenerator<String> idGenerator = mock(IdGenerator.class);
+ private final Compensable compensable = mock(Compensable.class);
private final OmegaContext omegaContext = new OmegaContext(idGenerator);
private final TransactionAspect aspect = new TransactionAspect(sender, omegaContext);
@@ -160,6 +168,23 @@ public void sendsAbortEventOnTimeout() throws Throwable {
assertThat(messages.size(), is(2));
}
+ @Test
+ public void returnImmediatelyWhenReceivedRejectResponse() throws Throwable {
+ MessageSender sender = mock(MessageSender.class);
+ when(sender.send(any())).thenReturn(new AlphaResponse(true));
+
+ TransactionAspect aspect = new TransactionAspect(sender, omegaContext);
+ try {
+ aspect.advise(joinPoint, compensable);
+ expectFailing(InvalidTransactionException.class);
+ } catch (InvalidTransactionException e) {
+ System.out.println(e.getMessage());
+ assertThat(e.getMessage().contains("Abort sub transaction"), is(true));
+ }
+
+ verify(sender, times(1)).send(any());
+ }
+
private String doNothing() {
return "doNothing";
}
diff --git a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
index 9522b6f0..26368814 100644
--- a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
+++ b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
@@ -33,6 +33,7 @@ message GrpcServiceConfig {
}
message GrpcAck {
+ bool aborted = 1;
}
message GrpcTxEvent {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services