You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by zh...@apache.org on 2019/07/03 02:51:14 UTC
[servicecomb-pack] branch SCB-1321 updated (a620a1f -> 00dc53c)
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a change to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git.
from a620a1f SCB-1321 Improve test cases for AlphaIntegrationFsmTest
new 421f960 SCB-1321 Add test case duplicateTxOnFailureEvents
new 482cf5e SCB-1321 Set endTime of SagaData after actor terminates
new 249fb5f SCB-1321 Record throwable stack to SagaData
new 00dc53c SCB-1321 await wait 2 seconds
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../servicecomb/pack/alpha/fsm/SagaActor.java | 30 +++++++------
.../pack/alpha/fsm/domain/UpdateTxEventDomain.java | 12 ++++-
.../pack/alpha/fsm/event/TxAbortedEvent.java | 14 ++++++
.../servicecomb/pack/alpha/fsm/model/TxEntity.java | 14 ++++++
.../pack/alpha/fsm/SagaIntegrationTest.java | 22 +++++-----
.../alpha/server/fsm/GrpcSagaEventService.java | 3 +-
.../alpha/server/fsm/AlphaIntegrationFsmTest.java | 51 +++++++++++++++++-----
.../alpha/server/fsm/OmegaEventSagaSimulator.java | 24 +++++++---
8 files changed, 126 insertions(+), 44 deletions(-)
[servicecomb-pack] 02/04: SCB-1321 Set endTime of SagaData after
actor terminates
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 482cf5e52ce98af8a7d2be58b37fffd75e913a8b
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Wed Jul 3 09:20:44 2019 +0800
SCB-1321 Set endTime of SagaData after actor terminates
---
.../apache/servicecomb/pack/alpha/fsm/SagaActor.java | 17 +++++++++--------
1 file changed, 9 insertions(+), 8 deletions(-)
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
index 72be072..fc5477c 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
@@ -256,8 +256,6 @@ public class SagaActor extends
}
).event(TxComponsitedCheckInternalEvent.class, SagaData.class,
(event, data) -> {
-// if ((!data.isTerminated() && data.getCompensationRunningCounter().intValue() > 0)
-// || hasCommittedTx(data)) {
if (hasCompensationSentTx(data)) {
return stay().replying(data);
} else {
@@ -268,7 +266,7 @@ public class SagaActor extends
}
).event(SagaAbortedEvent.class, SagaData.class,
(event, data) -> {
- data.setTerminated(true);
+ //data.setTerminated(true);
if (hasCommittedTx(data)) {
SagaEndedDomain domainEvent = new SagaEndedDomain(SagaActorState.FAILED);
return stay().replying(data).applying(domainEvent);
@@ -313,7 +311,6 @@ public class SagaActor extends
when(SagaActorState.COMMITTED,
matchAnyEvent(
(event, data) -> {
- data.setEndTime(System.currentTimeMillis());
/**
* deleteMessages 只会删除redis中actor的数据，但是不会删除actor的highestSequenceNr https://github.com/akka/akka/issues/21181
* 已停止的 actor highestSequenceNr 需要手动清理，例如 actor 的持久化ID为 3c500008-7b9f-415f-b2fd-e6ad0d455fc1
@@ -332,7 +329,6 @@ public class SagaActor extends
when(SagaActorState.SUSPENDED,
matchAnyEvent(
(event, data) -> {
- data.setEndTime(System.currentTimeMillis());
deleteMessages(lastSequenceNr());
deleteSnapshot(snapshotSequenceNr());
return stop();
@@ -343,7 +339,6 @@ public class SagaActor extends
when(SagaActorState.COMPENSATED,
matchAnyEvent(
(event, data) -> {
- data.setEndTime(System.currentTimeMillis());
deleteMessages(lastSequenceNr());
deleteSnapshot(snapshotSequenceNr());
return stop();
@@ -375,6 +370,7 @@ public class SagaActor extends
LOG.info("stop {} {}", data.getGlobalTxId(), state);
data.setTerminated(true);
data.setLastState(state);
+ data.setEndTime(System.currentTimeMillis());
SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(getContext().getSystem())
.putSagaData(data.getGlobalTxId(), data);
}
@@ -412,6 +408,9 @@ public class SagaActor extends
TxEntity txEntity = data.getTxEntityMap().get(domainEvent.getLocalTxId());
txEntity.setEndTime(System.currentTimeMillis());
if (domainEvent.getState() == TxState.COMMITTED) {
+ // stop
+ data.setEndTime(System.currentTimeMillis());
+ data.setTerminated(true);
txEntity.setState(domainEvent.getState());
} else if (domainEvent.getState() == TxState.FAILED) {
txEntity.setState(domainEvent.getState());
@@ -437,9 +436,11 @@ public class SagaActor extends
}
});
} else if (domainEvent.getState() == SagaActorState.SUSPENDED) {
-
+ data.setEndTime(System.currentTimeMillis());
+ data.setTerminated(true);
} else if (domainEvent.getState() == SagaActorState.COMPENSATED) {
-
+ data.setEndTime(System.currentTimeMillis());
+ data.setTerminated(true);
}
}
LOG.debug("applyEvent: {} {}", stateName(), stateData().getGlobalTxId());
[servicecomb-pack] 01/04: SCB-1321 Add test case
duplicateTxOnFailureEvents
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 421f96038121c444de35da13f0ad9643f223f114
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Wed Jul 3 09:18:30 2019 +0800
SCB-1321 Add test case duplicateTxOnFailureEvents
---
.../alpha/server/fsm/AlphaIntegrationFsmTest.java | 35 ++++++++++++++++++++--
.../alpha/server/fsm/OmegaEventSagaSimulator.java | 14 +++++++++
2 files changed, 46 insertions(+), 3 deletions(-)
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
index d4d3e99..b366c53 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
@@ -93,10 +93,14 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().sagaSuccessfulEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> omegaEventSender.getOmegaCallbacks() != null);
- await().atMost(1, SECONDS).until(() -> SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId) != null);
+ await().atMost(1, SECONDS).until(() -> {
+ SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
+ return sagaData !=null && sagaData.getLastState()==SagaActorState.COMMITTED;
+ });
+
+ //await().atMost(1, SECONDS).until(() -> SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId) != null);
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
- Assert.assertEquals(sagaData.getLastState(),SagaActorState.COMMITTED);
+ //Assert.assertEquals(sagaData.getLastState(),SagaActorState.COMMITTED);
Assert.assertEquals(sagaData.getTxEntityMap().size(),3);
Assert.assertTrue(sagaData.getBeginTime() > 0);
Assert.assertTrue(sagaData.getEndTime() > 0);
@@ -345,4 +349,29 @@ public class AlphaIntegrationFsmTest {
Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(),TxState.COMPENSATED);
Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(),TxState.FAILED);
}
+
+ @Test
+ public void doNotCompensateDuplicateTxOnFailure() {
+ omegaEventSender.onConnected();
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+ omegaEventSender.getOmegaEventSagaSimulator().duplicateTxOnFailureEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
+ omegaEventSender.getBlockingStub().onTxEvent(event);
+ });
+ await().atMost(1, SECONDS).until(() -> {
+ SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
+ return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
+ });
+ SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
+ Assert.assertEquals(sagaData.getLastState(),SagaActorState.COMPENSATED);
+ Assert.assertEquals(sagaData.getTxEntityMap().size(),3);
+ Assert.assertTrue(sagaData.getBeginTime() > 0);
+ Assert.assertTrue(sagaData.getEndTime() > 0);
+ Assert.assertTrue(sagaData.getEndTime() > sagaData.getBeginTime());
+ Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(),TxState.COMPENSATED);
+ Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(),TxState.COMPENSATED);
+ Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(),TxState.FAILED);
+ }
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
index 22b5cac..8623953 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
@@ -164,6 +164,20 @@ public class OmegaEventSagaSimulator {
return sagaEvents;
}
+ public List<GrpcTxEvent> duplicateTxOnFailureEvents(String globalTxId, String localTxId_1, String localTxId_2, String localTxId_3){
+ List<GrpcTxEvent> sagaEvents = new ArrayList<>();
+ sagaEvents.add(sagaStartedEvent(globalTxId));
+ sagaEvents.add(txStartedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
+ sagaEvents.add(txEndedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
+ sagaEvents.add(txEndedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
+ sagaEvents.add(txStartedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
+ sagaEvents.add(txEndedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
+ sagaEvents.add(txStartedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
+ sagaEvents.add(txAbortedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
+ sagaEvents.add(sagaAbortedEvent(globalTxId));
+ return sagaEvents;
+ }
+
private GrpcTxEvent sagaStartedEvent(String globalTxId) {
return eventOf(EventType.SagaStartedEvent, globalTxId, globalTxId,
null, new byte[0], "", 0, "",
[servicecomb-pack] 03/04: SCB-1321 Record throwable stack to
SagaData
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 249fb5f6ec6505e42fe86230de07aa2d1c9ed1b7
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Wed Jul 3 10:05:28 2019 +0800
SCB-1321 Record throwable stack to SagaData
---
.../org/apache/servicecomb/pack/alpha/fsm/SagaActor.java | 13 +++++++------
.../pack/alpha/fsm/domain/UpdateTxEventDomain.java | 12 +++++++++++-
.../servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java | 14 ++++++++++++++
.../apache/servicecomb/pack/alpha/fsm/model/TxEntity.java | 14 ++++++++++++++
.../pack/alpha/server/fsm/GrpcSagaEventService.java | 3 ++-
.../pack/alpha/server/fsm/AlphaIntegrationFsmTest.java | 6 ++----
.../pack/alpha/server/fsm/OmegaEventSagaSimulator.java | 12 ++++++------
7 files changed, 56 insertions(+), 18 deletions(-)
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
index fc5477c..51d6c52 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
@@ -125,7 +125,7 @@ public class SagaActor extends
matchEvent(TxEndedEvent.class, SagaData.class,
(event, data) -> {
UpdateTxEventDomain domainEvent = new UpdateTxEventDomain(event.getParentTxId(),
- event.getLocalTxId(), TxState.COMMITTED);
+ event.getLocalTxId(), TxState.COMMITTED, new byte[0]);
if (data.getExpirationTime() > 0) {
return goTo(SagaActorState.PARTIALLY_COMMITTED)
.applying(domainEvent)
@@ -163,7 +163,7 @@ public class SagaActor extends
).event(TxAbortedEvent.class,
(event, data) -> {
UpdateTxEventDomain domainEvent = new UpdateTxEventDomain(event.getParentTxId(),
- event.getLocalTxId(), TxState.FAILED);
+ event.getLocalTxId(), TxState.FAILED, event.getPayloads());
return goTo(SagaActorState.FAILED)
.applying(domainEvent);
}
@@ -195,7 +195,7 @@ public class SagaActor extends
).event(TxEndedEvent.class,
(event, data) -> {
UpdateTxEventDomain domainEvent = new UpdateTxEventDomain(event.getParentTxId(),
- event.getLocalTxId(), TxState.COMMITTED);
+ event.getLocalTxId(), TxState.COMMITTED, new byte[0]);
if (data.getExpirationTime() > 0) {
return stay()
.applying(domainEvent)
@@ -228,7 +228,7 @@ public class SagaActor extends
).event(TxAbortedEvent.class,
(event, data) -> {
UpdateTxEventDomain domainEvent = new UpdateTxEventDomain(event.getParentTxId(),
- event.getLocalTxId(), TxState.FAILED);
+ event.getLocalTxId(), TxState.FAILED, event.getPayloads());
return goTo(SagaActorState.FAILED).applying(domainEvent);
}
).event(Arrays.asList(StateTimeout()), SagaData.class,
@@ -249,7 +249,7 @@ public class SagaActor extends
).event(TxCompensatedEvent.class, SagaData.class,
(event, data) -> {
UpdateTxEventDomain domainEvent = new UpdateTxEventDomain(event.getParentTxId(),
- event.getLocalTxId(), TxState.COMPENSATED);
+ event.getLocalTxId(), TxState.COMPENSATED, new byte[0]);
return stay().applying(domainEvent).andThen(exec(_data -> {
self().tell(TxComponsitedCheckInternalEvent.builder().build(), self());
}));
@@ -295,7 +295,7 @@ public class SagaActor extends
).event(TxEndedEvent.class, SagaData.class,
(event, data) -> {
UpdateTxEventDomain domainEvent = new UpdateTxEventDomain(event.getParentTxId(),
- event.getLocalTxId(), TxState.COMMITTED);
+ event.getLocalTxId(), TxState.COMMITTED, new byte[0]);
return stay().applying(domainEvent).andThen(exec(_data -> {
TxEntity txEntity = _data.getTxEntityMap().get(event.getLocalTxId());
// call compensate
@@ -414,6 +414,7 @@ public class SagaActor extends
txEntity.setState(domainEvent.getState());
} else if (domainEvent.getState() == TxState.FAILED) {
txEntity.setState(domainEvent.getState());
+ txEntity.setThrowablePayLoads(domainEvent.getThrowablePayLoads());
data.getTxEntityMap().forEach((k, v) -> {
if (v.getState() == TxState.COMMITTED) {
// call compensate
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/UpdateTxEventDomain.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/UpdateTxEventDomain.java
index 839cfe0..2c8831c 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/UpdateTxEventDomain.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/UpdateTxEventDomain.java
@@ -23,11 +23,13 @@ public class UpdateTxEventDomain implements DomainEvent {
private String parentTxId;
private String localTxId;
private TxState state;
+ private byte[] throwablePayLoads;
- public UpdateTxEventDomain(String parentTxId, String localTxId, TxState state) {
+ public UpdateTxEventDomain(String parentTxId, String localTxId, TxState state, byte[] throwablePayLoads) {
this.parentTxId = parentTxId;
this.localTxId = localTxId;
this.state = state;
+ this.throwablePayLoads = throwablePayLoads;
}
public String getParentTxId() {
@@ -53,4 +55,12 @@ public class UpdateTxEventDomain implements DomainEvent {
public void setState(TxState state) {
this.state = state;
}
+
+ public byte[] getThrowablePayLoads() {
+ return throwablePayLoads;
+ }
+
+ public void setThrowablePayLoads(byte[] throwablePayLoads) {
+ this.throwablePayLoads = throwablePayLoads;
+ }
}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
index 8ee61de..ca1f9e4 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
@@ -21,6 +21,15 @@ import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
public class TxAbortedEvent extends TxEvent {
+ private byte[] payloads;
+
+ public byte[] getPayloads() {
+ return payloads;
+ }
+
+ public void setPayloads(byte[] payloads) {
+ this.payloads = payloads;
+ }
public static Builder builder() {
return new Builder();
@@ -49,6 +58,11 @@ public class TxAbortedEvent extends TxEvent {
return this;
}
+ public Builder payloads(byte[] payloads) {
+ txAbortedEvent.setPayloads(payloads);
+ return this;
+ }
+
public TxAbortedEvent build() {
return txAbortedEvent;
}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
index 0122adc..de849e2 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
@@ -31,6 +31,7 @@ public class TxEntity implements Serializable {
private TxState state;
private String compensationMethod;
private byte[] payloads;
+ private byte[] throwablePayLoads;
public String getServiceName() {
return serviceName;
@@ -112,6 +113,14 @@ public class TxEntity implements Serializable {
this.payloads = payloads;
}
+ public byte[] getThrowablePayLoads() {
+ return throwablePayLoads;
+ }
+
+ public void setThrowablePayLoads(byte[] throwablePayLoads) {
+ this.throwablePayLoads = throwablePayLoads;
+ }
+
public static Builder builder() {
return new Builder();
}
@@ -159,6 +168,11 @@ public class TxEntity implements Serializable {
return this;
}
+ public Builder throwablePayLoads(byte[] throwablePayLoads) {
+ txEntity.setThrowablePayLoads(throwablePayLoads);
+ return this;
+ }
+
public Builder state(TxState state) {
txEntity.setState(state);
return this;
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java
index a54ef83..2394ab2 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java
@@ -115,7 +115,8 @@ public class GrpcSagaEventService extends TxEventServiceImplBase {
event = org.apache.servicecomb.pack.alpha.fsm.event.TxAbortedEvent.builder()
.globalTxId(message.getGlobalTxId())
.parentTxId(message.getParentTxId())
- .localTxId(message.getLocalTxId()).build();
+ .localTxId(message.getLocalTxId())
+ .payloads(message.getPayloads().toByteArray()).build();
} else if (message.getType().equals(EventType.TxCompensatedEvent.name())) {
event = org.apache.servicecomb.pack.alpha.fsm.event.TxCompensatedEvent.builder()
.globalTxId(message.getGlobalTxId())
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
index b366c53..3080e2c 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
@@ -97,10 +97,7 @@ public class AlphaIntegrationFsmTest {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMMITTED;
});
-
- //await().atMost(1, SECONDS).until(() -> SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId) != null);
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
- //Assert.assertEquals(sagaData.getLastState(),SagaActorState.COMMITTED);
Assert.assertEquals(sagaData.getTxEntityMap().size(),3);
Assert.assertTrue(sagaData.getBeginTime() > 0);
Assert.assertTrue(sagaData.getEndTime() > 0);
@@ -164,7 +161,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().lastTxAbortedEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
});
@@ -177,6 +174,7 @@ public class AlphaIntegrationFsmTest {
Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(),TxState.COMPENSATED);
Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(),TxState.COMPENSATED);
Assert.assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(),TxState.FAILED);
+ Assert.assertArrayEquals(sagaData.getTxEntityMap().get(localTxId_3).getThrowablePayLoads(),NullPointerException.class.getName().getBytes());
}
@Test
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
index 8623953..52bcba5 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
@@ -45,7 +45,7 @@ public class OmegaEventSagaSimulator {
List<GrpcTxEvent> sagaEvents = new ArrayList<>();
sagaEvents.add(sagaStartedEvent(globalTxId));
sagaEvents.add(txStartedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
- sagaEvents.add(txAbortedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
+ sagaEvents.add(txAbortedEvent(globalTxId, localTxId_1, globalTxId, NullPointerException.class.getName().getBytes(), "method a"));
sagaEvents.add(sagaAbortedEvent(globalTxId));
return sagaEvents;
}
@@ -56,7 +56,7 @@ public class OmegaEventSagaSimulator {
sagaEvents.add(txStartedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
sagaEvents.add(txEndedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
sagaEvents.add(txStartedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
- sagaEvents.add(txAbortedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
+ sagaEvents.add(txAbortedEvent(globalTxId, localTxId_2, globalTxId, NullPointerException.class.getName().getBytes(), "method b"));
sagaEvents.add(sagaAbortedEvent(globalTxId));
return sagaEvents;
}
@@ -69,7 +69,7 @@ public class OmegaEventSagaSimulator {
sagaEvents.add(txStartedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
sagaEvents.add(txEndedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
sagaEvents.add(txStartedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
- sagaEvents.add(txAbortedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
+ sagaEvents.add(txAbortedEvent(globalTxId, localTxId_3, globalTxId, NullPointerException.class.getName().getBytes(), "method c"));
sagaEvents.add(sagaAbortedEvent(globalTxId));
return sagaEvents;
}
@@ -78,7 +78,7 @@ public class OmegaEventSagaSimulator {
List<GrpcTxEvent> sagaEvents = new ArrayList<>();
sagaEvents.add(sagaStartedEvent(globalTxId));
sagaEvents.add(txStartedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
- sagaEvents.add(txAbortedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
+ sagaEvents.add(txAbortedEvent(globalTxId, localTxId_1, globalTxId, NullPointerException.class.getName().getBytes(), "method a"));
sagaEvents.add(txStartedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
sagaEvents.add(txEndedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
sagaEvents.add(txStartedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
@@ -159,7 +159,7 @@ public class OmegaEventSagaSimulator {
sagaEvents.add(txStartedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
sagaEvents.add(txEndedEvent(globalTxId, localTxId_1, globalTxId, "service a".getBytes(), "method a"));
sagaEvents.add(txEndedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
- sagaEvents.add(txAbortedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
+ sagaEvents.add(txAbortedEvent(globalTxId, localTxId_3, globalTxId, NullPointerException.class.getName().getBytes(), "method c"));
sagaEvents.add(sagaAbortedEvent(globalTxId));
return sagaEvents;
}
@@ -173,7 +173,7 @@ public class OmegaEventSagaSimulator {
sagaEvents.add(txStartedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
sagaEvents.add(txEndedEvent(globalTxId, localTxId_2, globalTxId, "service b".getBytes(), "method b"));
sagaEvents.add(txStartedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
- sagaEvents.add(txAbortedEvent(globalTxId, localTxId_3, globalTxId, "service c".getBytes(), "method c"));
+ sagaEvents.add(txAbortedEvent(globalTxId, localTxId_3, globalTxId, NullPointerException.class.getName().getBytes(), "method c"));
sagaEvents.add(sagaAbortedEvent(globalTxId));
return sagaEvents;
}
[servicecomb-pack] 04/04: SCB-1321 await wait 2 seconds
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 00dc53c436a07b757b2f575c1d4d1cf41a694909
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Wed Jul 3 10:50:49 2019 +0800
SCB-1321 await wait 2 seconds
---
.../pack/alpha/fsm/SagaIntegrationTest.java | 22 +++++++++++-----------
.../alpha/server/fsm/AlphaIntegrationFsmTest.java | 20 ++++++++++----------
2 files changed, 21 insertions(+), 21 deletions(-)
diff --git a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaIntegrationTest.java b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaIntegrationTest.java
index 9a65bfb..7fbf843 100644
--- a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaIntegrationTest.java
+++ b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaIntegrationTest.java
@@ -57,7 +57,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMMITTED
@@ -81,7 +81,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMPENSATED
@@ -104,7 +104,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMPENSATED
@@ -129,7 +129,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMPENSATED
@@ -155,7 +155,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMPENSATED
@@ -181,7 +181,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMPENSATED
@@ -207,7 +207,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMPENSATED
@@ -233,7 +233,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.SUSPENDED
@@ -286,7 +286,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMMITTED
@@ -312,7 +312,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMMITTED
@@ -338,7 +338,7 @@ public class SagaIntegrationTest {
sagaEventBus.post(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
if(sagaData != null){
return sagaData.getLastState() == SagaActorState.COMPENSATED
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
index 3080e2c..27f5fd3 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/AlphaIntegrationFsmTest.java
@@ -93,7 +93,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().sagaSuccessfulEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMMITTED;
});
@@ -115,7 +115,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().firstTxAbortedEvents(globalTxId, localTxId_1).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
});
@@ -137,7 +137,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().middleTxAbortedEvents(globalTxId, localTxId_1, localTxId_2).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
});
@@ -187,7 +187,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().receivedRemainingEventAfterFirstTxAbortedEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
});
@@ -212,7 +212,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().sagaAbortedEventAfterAllTxEndedsEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
});
@@ -237,7 +237,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().omegaSendSagaTimeoutEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.SUSPENDED;
});
@@ -286,7 +286,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().successfulWithTxConcurrentEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMMITTED;
});
@@ -310,7 +310,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().successfulWithTxConcurrentCrossEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMMITTED;
});
@@ -334,7 +334,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().lastTxAbortedEventWithTxConcurrentEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
});
@@ -358,7 +358,7 @@ public class AlphaIntegrationFsmTest {
omegaEventSender.getOmegaEventSagaSimulator().duplicateTxOnFailureEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
omegaEventSender.getBlockingStub().onTxEvent(event);
});
- await().atMost(1, SECONDS).until(() -> {
+ await().atMost(2, SECONDS).until(() -> {
SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
return sagaData !=null && sagaData.getLastState()==SagaActorState.COMPENSATED;
});