You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by GitBox <gi...@apache.org> on 2019/01/11 23:26:57 UTC
[GitHub] WillemJiang closed pull request #378: Scb 1103
WillemJiang closed pull request #378: Scb 1103
URL: https://github.com/apache/servicecomb-pack/pull/378
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-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/GrpcTccEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/GrpcTccEventService.java
index d92c55ac..2c00932c 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/GrpcTccEventService.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/GrpcTccEventService.java
@@ -23,14 +23,7 @@
import org.apache.servicecomb.pack.alpha.server.tcc.callback.OmegaCallbacksRegistry;
import org.apache.servicecomb.pack.alpha.server.tcc.jpa.EventConverter;
import org.apache.servicecomb.pack.alpha.server.tcc.service.TccTxEventService;
-import org.apache.servicecomb.pack.contract.grpc.GrpcAck;
-import org.apache.servicecomb.pack.contract.grpc.GrpcServiceConfig;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinateCommand;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccParticipatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionEndedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionStartedEvent;
-import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc;
+import org.apache.servicecomb.pack.contract.grpc.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -65,11 +58,20 @@ public void onTccTransactionStarted(GrpcTccTransactionStartedEvent request, Stre
}
@Override
- public void participate(GrpcTccParticipatedEvent request, StreamObserver<GrpcAck> responseObserver) {
- LOG.info("Received participated event from service {} , global tx id: {}, local tx id: {}", request.getServiceName(),
+ public void onParticipationStarted(GrpcParticipationStartedEvent request, StreamObserver<GrpcAck> responseObserver) {
+ LOG.info("Received participation started event from service {} , global tx id: {}, local tx id: {}", request.getServiceName(),
request.getGlobalTxId(), request.getLocalTxId()) ;
responseObserver.onNext(
- tccTxEventService.onParticipatedEvent(EventConverter.convertToParticipatedEvent(request)) ? ALLOW : REJECT);
+ tccTxEventService.onParticipationStartedEvent(EventConverter.convertToParticipatedEvent(request)) ? ALLOW : REJECT);
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void onParticipationEnded(GrpcParticipationEndedEvent request, StreamObserver<GrpcAck> responseObserver) {
+ LOG.info("Received participation ended event from service {} , global tx id: {}, local tx id: {}", request.getServiceName(),
+ request.getGlobalTxId(), request.getLocalTxId()) ;
+ responseObserver.onNext(
+ tccTxEventService.onParticipationEndedEvent(EventConverter.convertToParticipatedEvent(request)) ? ALLOW : REJECT);
responseObserver.onCompleted();
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/EventConverter.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/EventConverter.java
index 179e70a3..ea946bb6 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/EventConverter.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/EventConverter.java
@@ -18,22 +18,32 @@
package org.apache.servicecomb.pack.alpha.server.tcc.jpa;
import org.apache.servicecomb.pack.common.TransactionStatus;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccParticipatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionEndedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionStartedEvent;
+import org.apache.servicecomb.pack.contract.grpc.*;
public class EventConverter {
- public static ParticipatedEvent convertToParticipatedEvent(GrpcTccParticipatedEvent request) {
+ public static ParticipatedEvent convertToParticipatedEvent(GrpcParticipationStartedEvent request) {
return new ParticipatedEvent(
request.getServiceName(),
request.getInstanceId(),
request.getGlobalTxId(),
request.getLocalTxId(),
request.getParentTxId(),
- request.getConfirmMethod(),
- request.getCancelMethod(),
+ "",
+ "",
+ ""
+ );
+ }
+
+ public static ParticipatedEvent convertToParticipatedEvent(GrpcParticipationEndedEvent request) {
+ return new ParticipatedEvent(
+ request.getServiceName(),
+ request.getInstanceId(),
+ request.getGlobalTxId(),
+ request.getLocalTxId(),
+ request.getParentTxId(),
+ "",
+ "",
request.getStatus()
);
}
@@ -83,14 +93,25 @@ public static TccTxEvent convertToTccTxEvent(GrpcTccTransactionEndedEvent event)
event.getStatus());
}
- public static TccTxEvent convertToTccTxEvent(GrpcTccParticipatedEvent event) {
+ public static TccTxEvent convertToTccTxEvent(GrpcParticipationStartedEvent event) {
+ return new TccTxEvent(event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ TccTxType.PARTICIPATED.name(),
+ toMethodInfo("", ""),
+ "");
+ }
+
+ public static TccTxEvent convertToTccTxEvent(GrpcParticipationEndedEvent event) {
return new TccTxEvent(event.getServiceName(),
event.getInstanceId(),
event.getGlobalTxId(),
event.getLocalTxId(),
event.getParentTxId(),
TccTxType.PARTICIPATED.name(),
- toMethodInfo(event.getCancelMethod(), event.getConfirmMethod()),
+ toMethodInfo("", ""),
event.getStatus());
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/ParticipatedEventRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/ParticipatedEventRepository.java
index c9f11061..35209d4f 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/ParticipatedEventRepository.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/ParticipatedEventRepository.java
@@ -19,8 +19,13 @@
import java.util.List;
import java.util.Optional;
+
+import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.CrudRepository;
+import org.springframework.data.repository.query.Param;
+
+import javax.transaction.Transactional;
public interface ParticipatedEventRepository extends CrudRepository<ParticipatedEvent, Long> {
@@ -29,4 +34,14 @@
@Query(value = "SELECT t FROM ParticipatedEvent AS t WHERE t.globalTxId = ?1 and t.localTxId = ?2")
Optional<ParticipatedEvent> findByUniqueKey(String globalTxId, String localTxId);
+
+// @Query("UPDATE ParticipatedEvent t SET t.status = ?3 WHERE t.globalTxId = ?1 and t.localTxId = ?2 limit 1")
+// void updateStatusByUniqueKey(String globalTxId, String localTxId, String status);
+
+ @Transactional
+ @Modifying(clearAutomatically = true)
+ @Query("UPDATE ParticipatedEvent t SET t.status = :status WHERE t.globalTxId = :globalTxId and t.localTxId = :localTxId")
+ void updateStatusByUniqueKey(@Param("globalTxId") String globalTxId,
+ @Param("localTxId") String localTxId,
+ @Param("status") String status);
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/TccTxEvent.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/TccTxEvent.java
index af06a0e3..52378388 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/TccTxEvent.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/jpa/TccTxEvent.java
@@ -122,6 +122,10 @@ public Date getLastModified() {
return lastModified;
}
+ public void setStatus(String status) {
+ this.status = status;
+ }
+
@Override
public String toString() {
return "TccTxEvent{" +
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/MemoryTxEventRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/MemoryTxEventRepository.java
index 91309110..3251d833 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/MemoryTxEventRepository.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/MemoryTxEventRepository.java
@@ -57,6 +57,19 @@ public void saveParticipatedEvent(ParticipatedEvent event) {
save(EventConverter.convertToTccTxEvent(event));
}
+ @Override
+ public void updateParticipatedEventStatus(ParticipatedEvent event) {
+// Set<TccTxEvent> events = tccEventMap.get(event.getGlobalTxId());
+// if (events != null) {
+// TccTxEvent ev = events.stream().filter(e ->
+// event.getServiceName().equals(e.getServiceName())
+// && "PARTICIPATED".equals(e.getTxType())
+// && event.getLocalTxId().equals(e.getLocalTxId())).findAny().get();
+// ev.setStatus(event.getStatus());
+// }
+ save(EventConverter.convertToTccTxEvent(event));
+ }
+
@Override
public void coordinated(TccTxEvent event) {
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/RDBTxEventRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/RDBTxEventRepository.java
index b48e984b..d6bf3cab 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/RDBTxEventRepository.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/RDBTxEventRepository.java
@@ -64,6 +64,14 @@ public void saveParticipatedEvent(ParticipatedEvent event) {
tccTxEventDBRepository.save(EventConverter.convertToTccTxEvent(event));
}
+ @Override
+ @Transactional
+ public void updateParticipatedEventStatus(ParticipatedEvent event) {
+ participatedEventRepository.updateStatusByUniqueKey(event.getGlobalTxId(), event.getLocalTxId(), event.getStatus());
+ // saveTccEventHere
+ tccTxEventDBRepository.save(EventConverter.convertToTccTxEvent(event));
+ }
+
@Override
@Transactional
public void coordinated(TccTxEvent event) {
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventRepository.java
index f7afccc8..f6478dc4 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventRepository.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventRepository.java
@@ -33,6 +33,8 @@
void saveParticipatedEvent(ParticipatedEvent event);
+ void updateParticipatedEventStatus(ParticipatedEvent event);
+
void coordinated(TccTxEvent event);
void save(TccTxEvent event);
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventService.java
index 99a60835..380ae911 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventService.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventService.java
@@ -63,8 +63,8 @@ public boolean onTccStartedEvent(GlobalTxEvent globalTxEvent) {
return true;
}
- public boolean onParticipatedEvent(ParticipatedEvent participatedEvent) {
- LOG.info("Registered Participated event, global tx: {}, local tx: {}, parent id: {}, "
+ public boolean onParticipationStartedEvent(ParticipatedEvent participatedEvent) {
+ LOG.info("Registered Participation started event, global tx: {}, local tx: {}, parent id: {}, "
+ "confirm: {}, cancel: {}, status: {}, service [{}] instanceId [{}]",
participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), participatedEvent.getParentTxId(),
participatedEvent.getConfirmMethod(), participatedEvent.getCancelMethod(), participatedEvent.getStatus(),
@@ -81,6 +81,24 @@ public boolean onParticipatedEvent(ParticipatedEvent participatedEvent) {
return true;
}
+ public boolean onParticipationEndedEvent(ParticipatedEvent participatedEvent) {
+ LOG.info("Registered Participation ended event, global tx: {}, local tx: {}, parent id: {}, "
+ + "confirm: {}, cancel: {}, status: {}, service [{}] instanceId [{}]",
+ participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), participatedEvent.getParentTxId(),
+ participatedEvent.getConfirmMethod(), participatedEvent.getCancelMethod(), participatedEvent.getStatus(),
+ participatedEvent.getServiceName(), participatedEvent.getInstanceId());
+ try {
+ if (tccTxEventRepository.findByUniqueKey(participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), TccTxType.PARTICIPATED).isPresent()) {
+ tccTxEventRepository.updateParticipatedEventStatus(participatedEvent);
+ }
+ } catch (Exception ex) {
+ LOG.warn("Add participateEvent triggered exception, globalTxId:{}, localTxId:{}, ",
+ participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), ex);
+ return false;
+ }
+ return true;
+ }
+
public boolean onTccEndedEvent(GlobalTxEvent globalTxEvent) {
LOG.info("Registered TccEnded event, global tx: {}, local tx: {}, parent id: {}, "
+ "txType: {}, service [{}] instanceId [{}]",
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/AlphaTccServerTestBase.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/AlphaTccServerTestBase.java
index e26b7875..fe798d1a 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/AlphaTccServerTestBase.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/AlphaTccServerTestBase.java
@@ -39,14 +39,7 @@
import org.apache.servicecomb.pack.alpha.server.tcc.jpa.EventConverter;
import org.apache.servicecomb.pack.alpha.server.tcc.service.TccTxEventRepository;
import org.apache.servicecomb.pack.common.TransactionStatus;
-import org.apache.servicecomb.pack.contract.grpc.GrpcAck;
-import org.apache.servicecomb.pack.contract.grpc.GrpcServiceConfig;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinateCommand;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccParticipatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionEndedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionStartedEvent;
-import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc;
+import org.apache.servicecomb.pack.contract.grpc.*;
import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc.TccEventServiceBlockingStub;
import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc.TccEventServiceStub;
import org.hamcrest.core.Is;
@@ -154,17 +147,17 @@ public void assertOnTransactionStart() {
public void assertOnParticipated() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
- blockingStub.participate(newParticipatedEvent("Succeed"));
- blockingStub.participate(newParticipatedEvent("Succeed"));
+ blockingStub.onParticipationStarted(newParticipationStartedEvent());
+ blockingStub.onParticipationEnded(newParticipationEndedEvent("Succeed"));
List<TccTxEvent> events = tccTxEventRepository.findByGlobalTxId(globalTxId).get();
- assertThat(events.size(), is(1));
- TccTxEvent event = events.iterator().next();
+ assertThat(events.size(), is(2));
+ TccTxEvent event = events.get(1); // skip the first ParticipationStartedEvent.
assertThat(event.getGlobalTxId(), is(globalTxId));
assertThat(event.getLocalTxId(), is(localTxId));
assertThat(event.getInstanceId(), is(instanceId));
assertThat(event.getServiceName(), is(serviceName));
- assertThat(EventConverter.getMethodName(event.getMethodInfo(), true), is(confirmMethod));
- assertThat(EventConverter.getMethodName(event.getMethodInfo(), false), is(cancelMethod));
+// assertThat(EventConverter.getMethodName(event.getMethodInfo(), true), is(confirmMethod));
+// assertThat(EventConverter.getMethodName(event.getMethodInfo(), false), is(cancelMethod));
assertThat(event.getStatus(), is("Succeed"));
}
@@ -173,13 +166,14 @@ public void assertOnTccTransactionSucceedEnded() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
blockingStub.onTccTransactionStarted(newTxStart());
- blockingStub.participate(newParticipatedEvent("Succeed"));
+ blockingStub.onParticipationStarted(newParticipationStartedEvent());
+ blockingStub.onParticipationEnded(newParticipationEndedEvent("Succeed"));
blockingStub.onTccTransactionEnded(newTxEnd("Succeed"));
await().atMost(2, SECONDS).until(() -> !receivedCommands.isEmpty());
assertThat(receivedCommands.size(), is(1));
GrpcTccCoordinateCommand command = receivedCommands.poll();
- assertThat(command.getMethod(), is("confirm"));
+// assertThat(command.getMethod(), is("confirm"));
assertThat(command.getGlobalTxId(), is(globalTxId));
assertThat(command.getServiceName(), is(serviceName));
@@ -192,13 +186,14 @@ public void assertOnTccTransactionFailedEnded() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
blockingStub.onTccTransactionStarted(newTxStart());
- blockingStub.participate(newParticipatedEvent("Succeed"));
+ blockingStub.onParticipationStarted(newParticipationStartedEvent());
+ blockingStub.onParticipationEnded(newParticipationEndedEvent("Succeed"));
blockingStub.onTccTransactionEnded(newTxEnd("Failed"));
await().atMost(2, SECONDS).until(() -> !receivedCommands.isEmpty());
assertThat(receivedCommands.size(), is(1));
GrpcTccCoordinateCommand command = receivedCommands.poll();
- assertThat(command.getMethod(), is("cancel"));
+// assertThat(command.getMethod(), is("cancel"));
assertThat(command.getGlobalTxId(), is(globalTxId));
assertThat(command.getServiceName(), is(serviceName));
assertThat(commandStreamObserver.isCompleted(), is(false));
@@ -211,7 +206,8 @@ public void assertOnCallbackNotExist() {
OmegaCallbacksRegistry.getRegistry().remove(serviceName);
blockingStub.onTccTransactionStarted(newTxStart());
- blockingStub.participate(newParticipatedEvent("Succeed"));
+ blockingStub.onParticipationStarted(newParticipationStartedEvent());
+ blockingStub.onParticipationEnded(newParticipationEndedEvent("Succeed"));
GrpcAck result = blockingStub.onTccTransactionEnded(newTxEnd("Succeed"));
assertThat(result.getAborted(), is(true));
}
@@ -223,7 +219,8 @@ public void assertOnCallbacksExecuteError() {
OmegaCallbacksRegistry.getRegistry().get(serviceName).put(instanceId, new GrpcOmegaTccCallback(null));
blockingStub.onTccTransactionStarted(newTxStart());
- blockingStub.participate(newParticipatedEvent("Succeed"));
+ blockingStub.onParticipationStarted(newParticipationStartedEvent());
+ blockingStub.onParticipationEnded(newParticipationEndedEvent("Succeed"));
GrpcAck result = blockingStub.onTccTransactionEnded(newTxEnd("Succeed"));
assertThat(result.getAborted(), is(true));
@@ -244,31 +241,39 @@ public void assertOnSwitchOtherCallbackInstance() {
OmegaCallbacksRegistry.getRegistry().get(serviceName).remove(instanceId);
blockingStub.onTccTransactionStarted(newTxStart());
- blockingStub.participate(newParticipatedEvent("Succeed"));
+ blockingStub.onParticipationStarted(newParticipationStartedEvent());
+ blockingStub.onParticipationEnded(newParticipationEndedEvent("Succeed"));
GrpcAck result = blockingStub.onTccTransactionEnded(newTxEnd("Succeed"));
await().atMost(2, SECONDS).until(() -> !receivedCommands.isEmpty());
assertThat(receivedCommands.size(), is(1));
GrpcTccCoordinateCommand command = receivedCommands.poll();
- assertThat(command.getMethod(), is("confirm"));
+// assertThat(command.getMethod(), is("confirm"));
assertThat(command.getGlobalTxId(), is(globalTxId));
assertThat(command.getServiceName(), is(serviceName));
assertThat(result.getAborted(), is(false));
}
- private GrpcTccParticipatedEvent newParticipatedEvent(String status) {
- return GrpcTccParticipatedEvent.newBuilder()
+ private GrpcParticipationStartedEvent newParticipationStartedEvent() {
+ return GrpcParticipationStartedEvent.newBuilder()
.setGlobalTxId(globalTxId)
.setLocalTxId(localTxId)
.setServiceName(serviceName)
.setInstanceId(instanceId)
- .setCancelMethod(cancelMethod)
- .setConfirmMethod(confirmMethod)
- .setStatus(status)
.build();
}
+ private GrpcParticipationEndedEvent newParticipationEndedEvent(String status) {
+ return GrpcParticipationEndedEvent.newBuilder()
+ .setGlobalTxId(globalTxId)
+ .setLocalTxId(localTxId)
+ .setServiceName(serviceName)
+ .setInstanceId(instanceId)
+ .setStatus(status)
+ .build();
+ }
+
private GrpcTccTransactionStartedEvent newTxStart() {
return GrpcTccTransactionStartedEvent.newBuilder()
.setGlobalTxId(globalTxId)
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/TccCallbackEngineTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/TccCallbackEngineTest.java
index 7491b1dd..7b8f79fb 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/TccCallbackEngineTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/TccCallbackEngineTest.java
@@ -101,8 +101,8 @@ public void sendCoordinateCommandAfterTccEnd() {
StreamObserver responseObserver = mock(StreamObserver.class);
OmegaCallbacksRegistry.register(serviceConfig, responseObserver);
- tccTxEventService.onParticipatedEvent(participatedEvent);
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
+ tccTxEventService.onParticipationEndedEvent(participatedEvent);
tccTxEventService.onTccEndedEvent(tccEndEvent);
@@ -115,7 +115,7 @@ public void sendCoordinateFailedForOmegaDown() throws InterruptedException {
doThrow(IllegalArgumentException.class).when(responseObserver).onNext(any());
OmegaCallbacksRegistry.register(serviceConfig, responseObserver);
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
boolean result = tccCallbackEngine.execute(tccEndEvent);
assertThat(result, is(false));
@@ -135,7 +135,7 @@ public void doRetryCoordinateTillOmegaReceived() throws InterruptedException {
doThrow(IllegalArgumentException.class).when(failedResponseObserver).onNext(any());
OmegaCallbacksRegistry.register(serviceConfig, failedResponseObserver);
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
boolean result = tccCallbackEngine.execute(tccEndEvent);
assertThat(result, is(false));
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTest.java
index 476bd0bd..912311e5 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTest.java
@@ -106,7 +106,7 @@ public void onlyCoordinateParticipatedEventOnce() {
OmegaCallbacksRegistry.register(serviceConfig, observer);
tccTxEventService.onTccStartedEvent(tccStartEvent);
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
tccTxEventService.onTccEndedEvent(tccEndEvent);
tccTxEventService.onCoordinatedEvent(coordinateEvent);
@@ -123,7 +123,7 @@ public void handleTimeoutGlobalTraction() throws InterruptedException {
OmegaCallbacksRegistry.register(serviceConfig, observer);
tccTxEventService.onTccStartedEvent(tccStartEvent);
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
Thread.sleep(3000l);
Date deadLine = new Date(System.currentTimeMillis() - SECONDS.toMillis(2));
@@ -150,7 +150,7 @@ public void clearUpCompletedTxFromGlobalTxTable() {
OmegaCallbacksRegistry.register(serviceConfig, observer);
tccTxEventService.onTccStartedEvent(tccStartEvent);
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
tccTxEventService.onTccEndedEvent(tccEndEvent);
tccTxEventService.onCoordinatedEvent(coordinateEvent);
@@ -170,14 +170,14 @@ public void clearUpCompletedTxFromGlobalTxTableMoreThanOne() {
// one global tx
tccTxEventService.onTccStartedEvent(tccStartEvent);
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
tccTxEventService.onTccEndedEvent(tccEndEvent);
tccTxEventService.onCoordinatedEvent(coordinateEvent);
// another global tx
String globalTxId_2 = uniquify("globalTxId");
tccTxEventService.onTccStartedEvent(newGlobalTxEvent(TccTxType.STARTED, globalTxId_2, TransactionStatus.Succeed));
- tccTxEventService.onParticipatedEvent(newParticipateEvent(globalTxId_2, TransactionStatus.Succeed));
+ tccTxEventService.onParticipationStartedEvent(newParticipateEvent(globalTxId_2, TransactionStatus.Succeed));
tccTxEventService.onTccEndedEvent(newGlobalTxEvent(TccTxType.ENDED, globalTxId_2, TransactionStatus.Succeed));
tccTxEventService.onCoordinatedEvent(newTccTxEvent(TccTxType.COORDINATED, globalTxId_2, TransactionStatus.Succeed));
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTransactionTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTransactionTest.java
index d3e52cc4..f7f5e6fc 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTransactionTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/tcc/service/TccTxEventServiceTransactionTest.java
@@ -103,7 +103,7 @@ public void rollbackAfterSaveTccTxEventDbFailure() {
Optional<List<GlobalTxEvent>> startEvents = globalTxEventRepository.findByGlobalTxId(globalTxId);
assertThat(startEvents.isPresent(), is(false));
- tccTxEventService.onParticipatedEvent(participatedEvent);
+ tccTxEventService.onParticipationStartedEvent(participatedEvent);
Optional<List<ParticipatedEvent>> participates = participatedEventRepository.findByGlobalTxId(globalTxId);
assertThat(participates.isPresent(), is(false));
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
index e4755425..deca2936 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
@@ -19,22 +19,13 @@
import io.grpc.ManagedChannel;
+import org.apache.servicecomb.pack.contract.grpc.*;
import org.apache.servicecomb.pack.omega.connector.grpc.core.LoadBalanceContext;
import org.apache.servicecomb.pack.omega.context.ServiceConfig;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageHandler;
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageSender;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcAck;
-import org.apache.servicecomb.pack.contract.grpc.GrpcServiceConfig;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccParticipatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionEndedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionStartedEvent;
-import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc.TccEventServiceBlockingStub;
import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc.TccEventServiceStub;
@@ -79,11 +70,17 @@ public String target() {
}
@Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
- GrpcAck grpcAck = tccBlockingEventService.participate(convertTo(participateEvent));
+ public AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent) {
+ GrpcAck grpcAck = tccBlockingEventService.onParticipationStarted(convertTo(participationStartedEvent));
return new AlphaResponse(grpcAck.getAborted());
}
+ @Override
+ public AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent) {
+ GrpcAck grpcAck = tccBlockingEventService.onParticipationEnded(convertTo(participationEndedEvent));
+ return new AlphaResponse(grpcAck.getAborted());
+ }
+
@Override
public AlphaResponse tccTransactionStart(TccStartedEvent tccStartEvent) {
GrpcAck grpcAck = tccBlockingEventService.onTccTransactionStarted(convertTo(tccStartEvent));
@@ -142,16 +139,23 @@ private GrpcTccTransactionEndedEvent convertTo(TccEndedEvent tccEndEvent) {
.build();
}
- private GrpcTccParticipatedEvent convertTo(ParticipatedEvent participateEvent) {
- return GrpcTccParticipatedEvent.newBuilder()
+ private GrpcParticipationStartedEvent convertTo(ParticipationStartedEvent participationStartedEvent) {
+ return GrpcParticipationStartedEvent.newBuilder()
.setServiceName(serviceConfig.getServiceName())
.setInstanceId(serviceConfig.getInstanceId())
- .setGlobalTxId(participateEvent.getGlobalTxId())
- .setLocalTxId(participateEvent.getLocalTxId())
- .setParentTxId(participateEvent.getParentTxId())
- .setCancelMethod(participateEvent.getCancelMethod())
- .setConfirmMethod(participateEvent.getConfirmMethod())
- .setStatus(participateEvent.getStatus().toString())
+ .setGlobalTxId(participationStartedEvent.getGlobalTxId())
+ .setLocalTxId(participationStartedEvent.getLocalTxId())
+ .setParentTxId(participationStartedEvent.getParentTxId())
.build();
}
+ private GrpcParticipationEndedEvent convertTo(ParticipationEndedEvent participationEndedEvent) {
+ return GrpcParticipationEndedEvent.newBuilder()
+ .setServiceName(serviceConfig.getServiceName())
+ .setInstanceId(serviceConfig.getInstanceId())
+ .setGlobalTxId(participationEndedEvent.getGlobalTxId())
+ .setLocalTxId(participationEndedEvent.getLocalTxId())
+ .setParentTxId(participationEndedEvent.getParentTxId())
+ .setStatus(participationEndedEvent.getStatus().toString())
+ .build();
+ }
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSender.java
index 6356b1e8..145d8c79 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSender.java
@@ -27,7 +27,8 @@
import org.apache.servicecomb.pack.omega.transaction.OmegaException;
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationEndedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationStartedEvent;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
@@ -39,19 +40,35 @@ public TccLoadBalanceSender(LoadBalanceContext loadContext,
}
@Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
+ public AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent) {
do {
final TccMessageSender messageSender = pickMessageSender();
- Optional<AlphaResponse> response = doGrpcSend(messageSender, participateEvent, new SenderExecutor<ParticipatedEvent>() {
+ Optional<AlphaResponse> response = doGrpcSend(messageSender, participationStartedEvent, new SenderExecutor<ParticipationStartedEvent>() {
@Override
- public AlphaResponse apply(ParticipatedEvent event) {
- return messageSender.participate(event);
+ public AlphaResponse apply(ParticipationStartedEvent event) {
+ return messageSender.participationStart(event);
}
});
if (response.isPresent()) return response.get();
} while (!Thread.currentThread().isInterrupted());
- throw new OmegaException("Failed to send event " + participateEvent + " due to interruption");
+ throw new OmegaException("Failed to send event " + participationStartedEvent + " due to interruption");
+ }
+
+ @Override
+ public AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent) {
+ do {
+ final TccMessageSender messageSender = pickMessageSender();
+ Optional<AlphaResponse> response = doGrpcSend(messageSender, participationEndedEvent, new SenderExecutor<ParticipationEndedEvent>() {
+ @Override
+ public AlphaResponse apply(ParticipationEndedEvent event) {
+ return messageSender.participationEnd(event);
+ }
+ });
+ if (response.isPresent()) return response.get();
+ } while (!Thread.currentThread().isInterrupted());
+
+ throw new OmegaException("Failed to send event " + participationEndedEvent + " due to interruption");
}
@Override
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSenderTest.java
index 0dcdb144..baa16c01 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/GrpcTccClientMessageSenderTest.java
@@ -24,20 +24,12 @@
import static org.mockito.Mockito.verify;
import org.apache.servicecomb.pack.common.TransactionStatus;
+import org.apache.servicecomb.pack.contract.grpc.*;
import org.apache.servicecomb.pack.omega.context.ServiceConfig;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageHandler;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcAck;
-import org.apache.servicecomb.pack.contract.grpc.GrpcServiceConfig;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinateCommand;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccParticipatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionEndedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionStartedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationStartedEvent;
import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc.TccEventServiceImplBase;
import org.junit.Before;
import org.junit.Rule;
@@ -63,8 +55,8 @@
private final String localTxId = uniquify("localTxId");
private final String parentTxId = uniquify("parentTxId");
private final String methodName = uniquify("methodName");
- private final String confirmMethod = uniquify("confirmMethod");
- private final String cancelMethod = uniquify("cancleMethod");
+// private final String confirmMethod = uniquify("confirmMethod");
+// private final String cancelMethod = uniquify("cancleMethod");
private final String serviceName = uniquify("serviceName");
private final ServiceConfig serviceConfig = new ServiceConfig(uniquify("Service"));
@@ -198,14 +190,14 @@ public void onTccTransactionEnded(GrpcTccTransactionEndedEvent request,
}
@Test
- public void serviceOnParticipateTest() {
+ public void serviceOnStartParticipateTest() {
- final GrpcTccParticipatedEvent[] requestCaptor = new GrpcTccParticipatedEvent[1];
- ParticipatedEvent event = new ParticipatedEvent(globalTxId,localTxId, parentTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed);
+ final GrpcParticipationStartedEvent[] requestCaptor = new GrpcParticipationStartedEvent[1];
+ ParticipationStartedEvent event = new ParticipationStartedEvent(globalTxId,localTxId, parentTxId);
TccEventServiceImplBase serviceImpl = new TccEventServiceImplBase() {
- public void participate(GrpcTccParticipatedEvent request,
+ public void onParticipationStarted(GrpcParticipationStartedEvent request,
StreamObserver<GrpcAck> responseObserver) {
requestCaptor[0] = request;
responseObserver.onNext(ack);
@@ -214,15 +206,40 @@ public void participate(GrpcTccParticipatedEvent request,
};
serviceRegistry.addService(serviceImpl);
- AlphaResponse response =service.participate(event);
+ AlphaResponse response =service.participationStart(event);
+
+ assertThat(requestCaptor[0].getServiceName(), is(serviceConfig.serviceName()));
+ assertThat(requestCaptor[0].getInstanceId(), is(serviceConfig.instanceId()));
+ assertThat(requestCaptor[0].getGlobalTxId(), is(globalTxId));
+ assertThat(requestCaptor[0].getLocalTxId(), is(localTxId));
+ assertThat(requestCaptor[0].getParentTxId(), is(parentTxId));
+ assertThat(response.aborted(), is(false));
+ }
+
+ @Test
+ public void serviceOnEndParticipateTest() {
+
+ final GrpcParticipationEndedEvent[] requestCaptor = new GrpcParticipationEndedEvent[1];
+ ParticipationEndedEvent event = new ParticipationEndedEvent(globalTxId,localTxId, parentTxId, TransactionStatus.Succeed);
+
+ TccEventServiceImplBase serviceImpl = new TccEventServiceImplBase() {
+
+ public void onParticipationEnded(GrpcParticipationEndedEvent request,
+ StreamObserver<GrpcAck> responseObserver) {
+ requestCaptor[0] = request;
+ responseObserver.onNext(ack);
+ responseObserver.onCompleted();
+ }
+ };
+
+ serviceRegistry.addService(serviceImpl);
+ AlphaResponse response =service.participationEnd(event);
assertThat(requestCaptor[0].getServiceName(), is(serviceConfig.serviceName()));
assertThat(requestCaptor[0].getInstanceId(), is(serviceConfig.instanceId()));
assertThat(requestCaptor[0].getGlobalTxId(), is(globalTxId));
assertThat(requestCaptor[0].getLocalTxId(), is(localTxId));
assertThat(requestCaptor[0].getParentTxId(), is(parentTxId));
- assertThat(requestCaptor[0].getCancelMethod(), is(cancelMethod));
- assertThat(requestCaptor[0].getConfirmMethod(), is(confirmMethod));
assertThat(requestCaptor[0].getStatus(), is(TransactionStatus.Succeed.toString()));
assertThat(response.aborted(), is(false));
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/MyTccEventServiceImpl.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/MyTccEventServiceImpl.java
index 1b5bb3b7..d3127097 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/MyTccEventServiceImpl.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/MyTccEventServiceImpl.java
@@ -22,14 +22,8 @@
import io.grpc.stub.StreamObserver;
import java.lang.invoke.MethodHandles;
import java.util.Queue;
-import org.apache.servicecomb.pack.contract.grpc.GrpcAck;
-import org.apache.servicecomb.pack.contract.grpc.GrpcServiceConfig;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinateCommand;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccCoordinatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccParticipatedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionEndedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccTransactionStartedEvent;
-import org.apache.servicecomb.pack.contract.grpc.TccEventServiceGrpc;
+
+import org.apache.servicecomb.pack.contract.grpc.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,8 +64,8 @@ public void onTccTransactionStarted(GrpcTccTransactionStartedEvent request,
}
@Override
- public void participate(GrpcTccParticipatedEvent request, StreamObserver<GrpcAck> responseObserver) {
- LOG.info("Received participated event from service {} , global tx id: {}, local tx id: {}",
+ public void onParticipationStarted(GrpcParticipationStartedEvent request, StreamObserver<GrpcAck> responseObserver) {
+ LOG.info("Received participation started event from service {} , global tx id: {}, local tx id: {}",
request.getServiceName(),
request.getGlobalTxId(), request.getLocalTxId());
events.offer(request);
@@ -80,6 +74,17 @@ public void participate(GrpcTccParticipatedEvent request, StreamObserver<GrpcAck
responseObserver.onCompleted();
}
+ @Override
+ public void onParticipationEnded(GrpcParticipationEndedEvent request, StreamObserver<GrpcAck> responseObserver) {
+ LOG.info("Received participation ended event from service {} , global tx id: {}, local tx id: {}",
+ request.getServiceName(),
+ request.getGlobalTxId(), request.getLocalTxId());
+ events.offer(request);
+ sleep();
+ responseObserver.onNext(ALLOW);
+ responseObserver.onCompleted();
+ }
+
@Override
public void onTccTransactionEnded(GrpcTccTransactionEndedEvent request, StreamObserver<GrpcAck> responseObserver) {
LOG.info("Received transaction end event, global tx id: {}", request.getGlobalTxId());
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
index 810948de..92ca9a95 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
@@ -43,6 +43,7 @@
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.apache.servicecomb.pack.common.TransactionStatus;
+import org.apache.servicecomb.pack.contract.grpc.GrpcParticipationStartedEvent;
import org.apache.servicecomb.pack.omega.connector.grpc.LoadBalanceSenderTestBase;
import org.apache.servicecomb.pack.omega.connector.grpc.AlphaClusterConfig;
import org.apache.servicecomb.pack.omega.connector.grpc.core.FastestSender;
@@ -57,10 +58,9 @@
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageHandler;
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationStartedEvent;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
-import org.apache.servicecomb.pack.contract.grpc.GrpcTccParticipatedEvent;
import org.hamcrest.core.Is;
import org.junit.After;
import org.junit.AfterClass;
@@ -88,7 +88,7 @@
private final ServiceConfig serviceConfig = new ServiceConfig(serviceName);
- private ParticipatedEvent participatedEvent;
+ private ParticipationStartedEvent participationStartedEvent;
private TccStartedEvent tccStartedEvent;
private TccEndedEvent tccEndedEvent;
private CoordinatedEvent coordinatedEvent;
@@ -130,7 +130,7 @@ public void setup() {
loadContext =
new LoadBalanceContextBuilder(TransactionType.TCC, clusterConfig, serviceConfig, 30, 4).build();
tccLoadBalanceSender = new TccLoadBalanceSender(loadContext, new FastestSender());
- participatedEvent = new ParticipatedEvent(globalTxId, localTxId, parentTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed);
+ participationStartedEvent = new ParticipationStartedEvent(globalTxId, localTxId, parentTxId);
tccStartedEvent = new TccStartedEvent(globalTxId, localTxId);
tccEndedEvent = new TccEndedEvent(globalTxId, localTxId, TransactionStatus.Succeed);
coordinatedEvent = new CoordinatedEvent(globalTxId, localTxId, parentTxId, methodName, TransactionStatus.Succeed);
@@ -162,19 +162,19 @@ public void participatedSucceed() {
TccMessageSender actualSender = tccLoadBalanceSender.pickMessageSender();
assertThat(actualSender.target(), is(expectSender.target()));
- AlphaResponse response = tccLoadBalanceSender.participate(participatedEvent);
+ AlphaResponse response = tccLoadBalanceSender.participationStart(participationStartedEvent);
assertThat(loadContext.getSenders().get(actualSender), greaterThan(0L));
assertThat(response.aborted(), is(false));
Integer expectPort = Integer.valueOf(expectSender.target().split(":")[1]);
- GrpcTccParticipatedEvent result = (GrpcTccParticipatedEvent) eventsMap.get(expectPort).poll();
+ GrpcParticipationStartedEvent result = (GrpcParticipationStartedEvent) eventsMap.get(expectPort).poll();
assertThat(result.getGlobalTxId(), is(globalTxId));
- assertThat(result.getCancelMethod(), is(cancelMethod));
- assertThat(result.getConfirmMethod(), is(confirmMethod));
+// assertThat(result.getCancelMethod(), is(cancelMethod));
+// assertThat(result.getConfirmMethod(), is(confirmMethod));
assertThat(result.getServiceName(), is(serviceName));
assertThat(result.getInstanceId(), is(serviceConfig.instanceId()));
assertThat(result.getParentTxId(), is(parentTxId));
- assertThat(result.getStatus(), is(TransactionStatus.Succeed.name()));
+// assertThat(result.getStatus(), is(TransactionStatus.Succeed.name()));
}
@Test
@@ -189,15 +189,15 @@ public Boolean call() {
assertThat((connected.get(8080).size() == 1 && connected.get(8090).size() == 1), is(true));
// due to 8090 is slow than 8080, so 8080 will be routed with 2 times.
- tccLoadBalanceSender.participate(participatedEvent);
- tccLoadBalanceSender.participate(participatedEvent);
- tccLoadBalanceSender.participate(participatedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
assertThat(eventsMap.get(8080).size(), is(2));
assertThat(eventsMap.get(8090).size(), is(1));
// when 8080 was shutdown, request will be routed to 8090 automatically.
servers.get(8080).shutdownNow();
- tccLoadBalanceSender.participate(participatedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
assertThat(eventsMap.get(8090).size(), is(2));
// when 8080 was recovery, it will be routed again.
@@ -208,7 +208,7 @@ public Boolean call() {
return connected.get(8080).size() == 3;
}
});
- tccLoadBalanceSender.participate(participatedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
assertThat(eventsMap.get(8080).size(), is(3));
}
@@ -228,7 +228,7 @@ public Boolean call() {
}
try {
- tccLoadBalanceSender.participate(participatedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
} catch (OmegaException ex) {
assertThat(ex.getMessage().endsWith("all alpha server is down."), is(true));
}
@@ -240,15 +240,15 @@ public Boolean call() {
@Test(expected = OmegaException.class)
public void participateFailedThenAbort() {
TccMessageSender failedSender = mock(GrpcTccClientMessageSender.class);
- doThrow(new OmegaException("omega exception")).when(failedSender).participate((ParticipatedEvent)any());
+ doThrow(new OmegaException("omega exception")).when(failedSender).participationStart((ParticipationStartedEvent)any());
TccMessageSender succeedSender = mock(GrpcTccClientMessageSender.class);
- when(succeedSender.participate((ParticipatedEvent) any())).thenReturn(new AlphaResponse(false));
+ when(succeedSender.participationStart((ParticipationStartedEvent) any())).thenReturn(new AlphaResponse(false));
Map<MessageSender, Long> senders = Maps.newConcurrentMap();
senders.put(failedSender, 0l);
senders.put(succeedSender, 10l);
loadContext.setSenders(senders);
- tccLoadBalanceSender.participate(participatedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
}
@Test
@@ -258,7 +258,7 @@ public void participateInterruptedFailed() throws InterruptedException {
public void run() {
try {
await().atLeast(1, SECONDS);
- tccLoadBalanceSender.participate(participatedEvent);
+ tccLoadBalanceSender.participationStart(participationStartedEvent);
} catch (OmegaException e) {
assertThat(e.getMessage().endsWith("interruption"), Is.is(true));
}
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/MessageConfig.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/MessageConfig.java
index 415e42e9..d08be218 100644
--- a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/MessageConfig.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/MessageConfig.java
@@ -28,10 +28,7 @@
import org.apache.servicecomb.pack.omega.transaction.tcc.DefaultParametersContext;
import org.apache.servicecomb.pack.omega.transaction.tcc.ParametersContext;
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageSender;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
import org.mockito.Mockito;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -123,8 +120,14 @@ public String target() {
}
@Override
- public AlphaResponse participate(ParticipatedEvent participatedEvent) {
- messages.add(participatedEvent.toString());
+ public AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent) {
+ messages.add(participationStartedEvent.toString());
+ return new AlphaResponse(false);
+ }
+
+ @Override
+ public AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent) {
+ messages.add(participationEndedEvent.toString());
return new AlphaResponse(false);
}
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TccInterceptorTest.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TccInterceptorTest.java
index 5bda4ce5..6d28bc2f 100644
--- a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TccInterceptorTest.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TccInterceptorTest.java
@@ -31,10 +31,7 @@
import org.apache.servicecomb.pack.common.TransactionStatus;
import org.apache.servicecomb.pack.omega.context.IdGenerator;
import org.apache.servicecomb.pack.omega.transaction.tcc.TccMessageHandler;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -106,8 +103,10 @@ public void tccWorkflowSucceed() {
assertArrayEquals(
new String[] {
new TccStartedEvent(globalTxId, globalTxId).toString(),
- new ParticipatedEvent(globalTxId, newLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed).toString(),
- new ParticipatedEvent(globalTxId, anotherLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed).toString(),
+ new ParticipationStartedEvent(globalTxId, newLocalTxId, globalTxId).toString(),
+ new ParticipationEndedEvent(globalTxId, newLocalTxId, globalTxId, TransactionStatus.Succeed).toString(),
+ new ParticipationStartedEvent(globalTxId, anotherLocalTxId, globalTxId).toString(),
+ new ParticipationEndedEvent(globalTxId, anotherLocalTxId, globalTxId, TransactionStatus.Succeed).toString(),
new TccEndedEvent(globalTxId, globalTxId, TransactionStatus.Succeed).toString(),
new CoordinatedEvent(globalTxId, newLocalTxId, globalTxId, confirmMethod, TransactionStatus.Succeed).toString(),
new CoordinatedEvent(globalTxId, anotherLocalTxId, globalTxId, confirmMethod, TransactionStatus.Succeed).toString()
@@ -137,8 +136,10 @@ public void tccWorkflowFailed() {
assertArrayEquals(
new String[] {
new TccStartedEvent(globalTxId, globalTxId).toString(),
- new ParticipatedEvent(globalTxId, newLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed).toString(),
- new ParticipatedEvent(globalTxId, anotherLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Failed).toString(),
+ new ParticipationStartedEvent(globalTxId, newLocalTxId, globalTxId).toString(),
+ new ParticipationEndedEvent(globalTxId, newLocalTxId, globalTxId, TransactionStatus.Succeed).toString(),
+ new ParticipationStartedEvent(globalTxId, anotherLocalTxId, globalTxId).toString(),
+ new ParticipationEndedEvent(globalTxId, anotherLocalTxId, globalTxId, TransactionStatus.Failed).toString(),
new TccEndedEvent(globalTxId, globalTxId, TransactionStatus.Failed).toString(),
new CoordinatedEvent(globalTxId, newLocalTxId, globalTxId, cancelMethod, TransactionStatus.Succeed).toString()
},
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccEventService.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccEventService.java
index 370fe314..69f48592 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccEventService.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccEventService.java
@@ -18,7 +18,7 @@
package org.apache.servicecomb.pack.omega.transaction.tcc;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationStartedEvent;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
@@ -33,7 +33,7 @@
String target();
- AlphaResponse participate(ParticipatedEvent participatedEvent);
+ AlphaResponse participate(ParticipationStartedEvent participationStartedEvent);
AlphaResponse tccTransactionStart(TccStartedEvent tccStartedEvent);
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccMessageSender.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccMessageSender.java
index a1d16a80..bf299b0c 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccMessageSender.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccMessageSender.java
@@ -18,7 +18,8 @@
package org.apache.servicecomb.pack.omega.transaction.tcc;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationEndedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationStartedEvent;
import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
import org.apache.servicecomb.pack.omega.transaction.MessageSender;
@@ -26,7 +27,9 @@
public interface TccMessageSender extends MessageSender {
- AlphaResponse participate(ParticipatedEvent participateEvent);
+ AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent);
+
+ AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent);
AlphaResponse tccTransactionStart(TccStartedEvent tccStartEvent);
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspect.java
index f70f61a1..001a36e4 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspect.java
@@ -22,7 +22,8 @@
import org.apache.servicecomb.pack.common.TransactionStatus;
import org.apache.servicecomb.pack.omega.context.OmegaContext;
import org.apache.servicecomb.pack.omega.transaction.annotations.Participate;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationEndedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipationStartedEvent;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
@@ -58,18 +59,17 @@ Object advise(ProceedingJoinPoint joinPoint, Participate participate) throws Thr
LOG.debug("Updated context {} for participate method {} ", context, method.toString());
try {
+ tccMessageSender.participationStart(new ParticipationStartedEvent(context.globalTxId(), context.localTxId(), localTxId));
Object result = joinPoint.proceed();
// Send the participate message back
- tccMessageSender.participate(new ParticipatedEvent(context.globalTxId(), context.localTxId(), localTxId, confirmMethod,
- cancelMethod, TransactionStatus.Succeed));
+ tccMessageSender.participationEnd(new ParticipationEndedEvent(context.globalTxId(), context.localTxId(), localTxId, TransactionStatus.Succeed));
// Just store the parameters into the context
parametersContext.putParamters(context.localTxId(), joinPoint.getArgs());
LOG.debug("Participate Transaction with context {} has finished.", context);
return result;
} catch (Throwable throwable) {
// Now we don't handle the error message
- tccMessageSender.participate(new ParticipatedEvent(context.globalTxId(), context.localTxId(), localTxId, confirmMethod,
- cancelMethod, TransactionStatus.Failed));
+ tccMessageSender.participationEnd(new ParticipationEndedEvent(context.globalTxId(), context.localTxId(), localTxId, TransactionStatus.Failed));
LOG.error("Participate Transaction with context {} failed.", context, throwable);
throw throwable;
} finally {
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipationEndedEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipationEndedEvent.java
new file mode 100644
index 00000000..3d37ff29
--- /dev/null
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipationEndedEvent.java
@@ -0,0 +1,60 @@
+/*
+ * 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.pack.omega.transaction.tcc.events;
+
+import org.apache.servicecomb.pack.common.TransactionStatus;
+
+public class ParticipationEndedEvent
+{
+ private final String globalTxId;
+ private final String localTxId;
+ private final String parentTxId;
+ private final TransactionStatus status;
+
+ public ParticipationEndedEvent(String globalTxId, String localTxId, String parentTxId, TransactionStatus status) {
+ this.globalTxId = globalTxId;
+ this.localTxId = localTxId;
+ this.parentTxId = parentTxId;
+ this.status = status;
+ }
+
+ public String getGlobalTxId() {
+ return globalTxId;
+ }
+
+ public String getLocalTxId() {
+ return localTxId;
+ }
+
+ public String getParentTxId() {
+ return parentTxId;
+ }
+
+ public TransactionStatus getStatus() {
+ return status;
+ }
+
+ @Override
+ public String toString() {
+ return "ParticipationStartedEvent{" +
+ "globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ ", parentTxId='" + parentTxId + '\'' +
+ ", status=" + status +
+ '}';
+ }
+}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipatedEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipationStartedEvent.java
similarity index 66%
rename from omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipatedEvent.java
rename to omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipationStartedEvent.java
index 701509d3..4a560420 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipatedEvent.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/ParticipationStartedEvent.java
@@ -19,24 +19,17 @@
import org.apache.servicecomb.pack.common.TransactionStatus;
-public class ParticipatedEvent {
+public class ParticipationStartedEvent {
private final String globalTxId;
private final String localTxId;
private final String parentTxId;
- private final String confirmMethod;
- private final String cancelMethod;
- private final TransactionStatus status;
- public ParticipatedEvent(String globalTxId, String localTxId, String parentTxId, String confirmMethod,
- String cancelMethod, TransactionStatus status) {
+ public ParticipationStartedEvent(String globalTxId, String localTxId, String parentTxId) {
this.globalTxId = globalTxId;
this.localTxId = localTxId;
this.parentTxId = parentTxId;
- this.confirmMethod = confirmMethod;
- this.cancelMethod = cancelMethod;
- this.status = status;
}
public String getGlobalTxId() {
@@ -51,27 +44,12 @@ public String getParentTxId() {
return parentTxId;
}
- public String getConfirmMethod() {
- return confirmMethod;
- }
-
- public String getCancelMethod() {
- return cancelMethod;
- }
-
- public TransactionStatus getStatus() {
- return status;
- }
-
@Override
public String toString() {
- return "ParticipatedEvent{" +
+ return "ParticipationStartedEvent{" +
"globalTxId='" + globalTxId + '\'' +
", localTxId='" + localTxId + '\'' +
", parentTxId='" + parentTxId + '\'' +
- ", confirmMethod='" + confirmMethod + '\'' +
- ", cancelMethod='" + cancelMethod + '\'' +
- ", status=" + status +
'}';
}
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/CoordinateMessageHandlerTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/CoordinateMessageHandlerTest.java
index ca2879e2..5f72fb0d 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/CoordinateMessageHandlerTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/CoordinateMessageHandlerTest.java
@@ -28,10 +28,7 @@
import org.apache.servicecomb.pack.omega.context.CallbackContext;
import org.apache.servicecomb.pack.omega.context.OmegaContext;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
@@ -61,7 +58,12 @@ public String target() {
}
@Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
+ public AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent) {
+ return null;
+ }
+
+ @Override
+ public AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent) {
return null;
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspectTest.java
index da194915..3c510a03 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccParticipatorAspectTest.java
@@ -31,10 +31,7 @@
import org.apache.servicecomb.pack.omega.context.OmegaContext;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
import org.apache.servicecomb.pack.omega.transaction.annotations.Participate;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.junit.Before;
@@ -48,7 +45,8 @@
private final String localTxId = UUID.randomUUID().toString();
private final String newLocalTxId = UUID.randomUUID().toString();
- private final List<ParticipatedEvent> participatedEvents = new ArrayList<>();
+ private final List<ParticipationStartedEvent> participationStartedEvents = new ArrayList<>();
+ private final List<ParticipationEndedEvent> participationEndedEvents = new ArrayList<>();
private final AlphaResponse response = new AlphaResponse(false);
private String confirmMethod;
private String cancelMethod;
@@ -74,8 +72,14 @@ public String target() {
}
@Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
- participatedEvents.add(participateEvent);
+ public AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent) {
+ participationStartedEvents.add(participationStartedEvent);
+ return response;
+ }
+
+ @Override
+ public AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent) {
+ participationEndedEvents.add(participationEndedEvent);
return response;
}
@@ -127,15 +131,15 @@ public void setUp() throws Exception {
public void participateMethodIsCalledSuccessed() throws Throwable {
aspect.advise(joinPoint, participate);
- assertThat(participatedEvents.size(), is(1));
- ParticipatedEvent participatedEvent = participatedEvents.get(0);
+ assertThat(participationStartedEvents.size(), is(1));
+ ParticipationStartedEvent participationStartedEvent = participationStartedEvents.get(0);
- assertThat(participatedEvent.getGlobalTxId(), is(globalTxId));
- assertThat(participatedEvent.getParentTxId(), is(localTxId));
- assertThat(participatedEvent.getLocalTxId(), is(newLocalTxId));
- assertThat(participatedEvent.getStatus(), is(TransactionStatus.Succeed));
- assertThat(participatedEvent.getCancelMethod(), is(cancelMethod));
- assertThat(participatedEvent.getConfirmMethod(), is(confirmMethod));
+ assertThat(participationStartedEvent.getGlobalTxId(), is(globalTxId));
+ assertThat(participationStartedEvent.getParentTxId(), is(localTxId));
+ assertThat(participationStartedEvent.getLocalTxId(), is(newLocalTxId));
+// assertThat(participationStartedEvent.getStatus(), is(TransactionStatus.Succeed));
+// assertThat(participationStartedEvent.getCancelMethod(), is(cancelMethod));
+// assertThat(participationStartedEvent.getConfirmMethod(), is(confirmMethod));
assertThat(omegaContext.globalTxId(), is(globalTxId));
assertThat(omegaContext.localTxId(), is(localTxId));
@@ -154,15 +158,15 @@ public void participateMethodIsCalledFailed() throws Throwable {
assertThat(e, is(oops));
}
- assertThat(participatedEvents.size(), is(1));
- ParticipatedEvent participatedEvent = participatedEvents.get(0);
+ assertThat(participationStartedEvents.size(), is(1));
+ ParticipationStartedEvent participationStartedEvent = participationStartedEvents.get(0);
- assertThat(participatedEvent.getGlobalTxId(), is(globalTxId));
- assertThat(participatedEvent.getParentTxId(), is(localTxId));
- assertThat(participatedEvent.getLocalTxId(), is(newLocalTxId));
- assertThat(participatedEvent.getStatus(), is(TransactionStatus.Failed));
- assertThat(participatedEvent.getCancelMethod(), is(cancelMethod));
- assertThat(participatedEvent.getConfirmMethod(), is(confirmMethod));
+ assertThat(participationStartedEvent.getGlobalTxId(), is(globalTxId));
+ assertThat(participationStartedEvent.getParentTxId(), is(localTxId));
+ assertThat(participationStartedEvent.getLocalTxId(), is(newLocalTxId));
+// assertThat(participationStartedEvent.getStatus(), is(TransactionStatus.Failed));
+// assertThat(participationStartedEvent.getCancelMethod(), is(cancelMethod));
+// assertThat(participationStartedEvent.getConfirmMethod(), is(confirmMethod));
assertThat(omegaContext.globalTxId(), is(globalTxId));
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAnnotationProcessorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAnnotationProcessorTest.java
index 96d0de6c..560f6525 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAnnotationProcessorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAnnotationProcessorTest.java
@@ -33,10 +33,7 @@
import org.apache.servicecomb.pack.omega.context.OmegaContext;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
import org.apache.servicecomb.pack.omega.transaction.OmegaException;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
import org.hamcrest.core.Is;
import org.junit.Assert;
import org.junit.Before;
@@ -75,7 +72,12 @@ public String target() {
}
@Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
+ public AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent) {
+ return null;
+ }
+
+ @Override
+ public AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent) {
return null;
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAspectTest.java
index 2572c72b..83f2827c 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/tcc/TccStartAspectTest.java
@@ -32,10 +32,7 @@
import org.apache.servicecomb.pack.omega.context.OmegaContext;
import org.apache.servicecomb.pack.omega.context.annotations.TccStart;
import org.apache.servicecomb.pack.omega.transaction.AlphaResponse;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.pack.omega.transaction.tcc.events.TccStartedEvent;
+import org.apache.servicecomb.pack.omega.transaction.tcc.events.*;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.junit.Before;
@@ -71,7 +68,12 @@ public String target() {
}
@Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
+ public AlphaResponse participationStart(ParticipationStartedEvent participationStartedEvent) {
+ return null;
+ }
+
+ @Override
+ public AlphaResponse participationEnd(ParticipationEndedEvent participationEndedEvent) {
return null;
}
diff --git a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTccEvent.proto b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTccEvent.proto
index 20edad45..609bf0af 100644
--- a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTccEvent.proto
+++ b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTccEvent.proto
@@ -26,7 +26,8 @@ import "GrpcCommon.proto";
service TccEventService {
rpc OnConnected (GrpcServiceConfig) returns (stream GrpcTccCoordinateCommand) {
}
- rpc participate(GrpcTccParticipatedEvent) returns (GrpcAck) {}
+ rpc OnParticipationStarted(GrpcParticipationStartedEvent) returns (GrpcAck) {}
+ rpc OnParticipationEnded(GrpcParticipationEndedEvent) returns (GrpcAck) {}
rpc OnTccTransactionStarted (GrpcTccTransactionStartedEvent) returns (GrpcAck) {}
rpc OnTccTransactionEnded (GrpcTccTransactionEndedEvent) returns (GrpcAck) {}
rpc OnTccCoordinated(GrpcTccCoordinatedEvent) returns(GrpcAck) {}
@@ -43,15 +44,22 @@ message GrpcTccTransactionStartedEvent {
string instanceId = 6;
}
-message GrpcTccParticipatedEvent {
+message GrpcParticipationStartedEvent {
+ int64 timestamp = 1;
+ string globalTxId = 2;
+ string localTxId = 3;
+ string parentTxId = 4;
+ string serviceName = 5;
+ string instanceId = 6;
+}
+
+message GrpcParticipationEndedEvent {
int64 timestamp = 1;
string globalTxId = 2;
string localTxId = 3;
string parentTxId = 4;
string serviceName = 5;
string instanceId = 6;
- string confirmMethod = 7;
- string cancelMethod = 8;
string status = 9;
}
----------------------------------------------------------------
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