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/05 06:25:11 UTC
[servicecomb-pack] 02/02: SCB-1321 Add test case for actor
compensation retry
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 589ccbefd73c3dad864469e7239f0f61bd0f7ced
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Fri Jul 5 14:22:44 2019 +0800
SCB-1321 Add test case for actor compensation retry
---
.../servicecomb/pack/alpha/fsm/SagaActor.java | 56 +++++++++++++++---
.../pack/alpha/fsm/domain/AddTxEventDomain.java | 12 +++-
.../fsm/event/consumer/SagaEventConsumer.java | 24 ++++++--
.../servicecomb/pack/alpha/fsm/model/TxEntity.java | 20 +++++++
.../integration/akka/SpringAkkaExtension.java | 5 +-
.../alpha/server/fsm/AlphaIntegrationFsmTest.java | 67 ++++++++++++++++++++++
.../pack/alpha/server/fsm/OmegaEventSender.java | 4 +-
7 files changed, 169 insertions(+), 19 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 5104aca..712a5f4 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
@@ -22,6 +22,7 @@ import akka.persistence.fsm.AbstractPersistentFSM;
import java.lang.invoke.MethodHandles;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
+import org.apache.servicecomb.pack.alpha.core.AlphaException;
import org.apache.servicecomb.pack.alpha.fsm.domain.AddTxEventDomain;
import org.apache.servicecomb.pack.alpha.fsm.domain.DomainEvent;
import org.apache.servicecomb.pack.alpha.fsm.domain.SagaEndedDomain;
@@ -88,7 +89,8 @@ public class SagaActor extends
event.getParentTxId(),
event.getLocalTxId(),
event.getPayloads(),
- event.getCompensationMethod());
+ event.getCompensationMethod(),
+ event.getRetries());
if (data.getExpirationTime() > 0) {
return goTo(SagaActorState.PARTIALLY_ACTIVE)
.applying(domainEvent)
@@ -143,7 +145,8 @@ public class SagaActor extends
event.getParentTxId(),
event.getLocalTxId(),
event.getPayloads(),
- event.getCompensationMethod());
+ event.getCompensationMethod(),
+ event.getRetries());
if (data.getExpirationTime() > 0) {
return stay()
.applying(domainEvent)
@@ -182,7 +185,8 @@ public class SagaActor extends
event.getParentTxId(),
event.getLocalTxId(),
event.getPayloads(),
- event.getCompensationMethod());
+ event.getCompensationMethod(),
+ event.getRetries());
if (data.getExpirationTime() > 0) {
return goTo(SagaActorState.PARTIALLY_ACTIVE)
.applying(domainEvent)
@@ -260,7 +264,10 @@ public class SagaActor extends
//if (hasCompensationSentTx(data)) {
return stay().replying(data);
} else {
+ SagaEndedDomain domainEvent = new SagaEndedDomain(
+ SagaActorState.COMPENSATED);
return goTo(SagaActorState.COMPENSATED)
+ .applying(domainEvent)
.replying(data)
.forMax(Duration.create(1, TimeUnit.MILLISECONDS));
}
@@ -290,7 +297,8 @@ public class SagaActor extends
event.getParentTxId(),
event.getLocalTxId(),
event.getPayloads(),
- event.getCompensationMethod());
+ event.getCompensationMethod(),
+ event.getRetries());
return stay().applying(domainEvent);
}
).event(TxEndedEvent.class, SagaData.class,
@@ -444,6 +452,9 @@ public class SagaActor extends
} else if (domainEvent.getState() == SagaActorState.COMPENSATED) {
data.setEndTime(System.currentTimeMillis());
data.setTerminated(true);
+ } else if (domainEvent.getState() == SagaActorState.COMMITTED) {
+ data.setEndTime(System.currentTimeMillis());
+ data.setTerminated(true);
}
}
LOG.debug("applyEvent: {} {}", stateName(), stateData().getGlobalTxId());
@@ -478,12 +489,43 @@ public class SagaActor extends
.count() > 0;
}
+ //call omega compensate method
private void compensation(TxEntity txEntity, SagaData data) {
// increments the compensation running counter by one
data.getCompensationRunningCounter().incrementAndGet();
- //TODO call omega compensate method
- SpringAkkaExtension.SPRING_EXTENSION_PROVIDER.get(context().system()).compensate(txEntity);
- LOG.info("compensate {}", txEntity.getLocalTxId());
txEntity.setState(TxState.COMPENSATION_SENT);
+ try{
+ SpringAkkaExtension.SPRING_EXTENSION_PROVIDER.get(context().system()).compensate(txEntity);
+ LOG.info("compensate {}", txEntity.getLocalTxId());
+ }catch (AlphaException ex){
+ LOG.error(ex.getMessage(),ex);
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ LOG.error(e.getMessage(),e);
+ }
+ compensation(txEntity,data);
+ }catch (Exception ex){
+ LOG.error("compensation failed "+txEntity.getLocalTxId(), ex);
+ if(txEntity.getRetries() > 0){
+ // which means the retry number
+ if(txEntity.getRetriesCounter().incrementAndGet() < txEntity.getRetries()){
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ LOG.error(e.getMessage(),e);
+ }
+ compensation(txEntity,data);
+ }
+ } else if(txEntity.getRetries() == -1){
+ // which means retry it until succeed
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ LOG.error(e.getMessage(),e);
+ }
+ compensation(txEntity,data);
+ }
+ }
}
}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java
index af5936d..520b7db 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java
@@ -25,16 +25,18 @@ public class AddTxEventDomain implements DomainEvent {
private String parentTxId;
private String localTxId;
private TxState state = TxState.ACTIVE;
+ private int retries;
private String compensationMethod;
private byte[] payloads;
- public AddTxEventDomain(String serviceName, String instanceId, String parentTxId, String localTxId, byte[] payloads, String compensationMethod) {
+ public AddTxEventDomain(String serviceName, String instanceId, String parentTxId, String localTxId, byte[] payloads, String compensationMethod, int retries) {
this.serviceName = serviceName;
this.instanceId = instanceId;
this.parentTxId = parentTxId;
this.localTxId = localTxId;
this.compensationMethod = compensationMethod;
this.payloads = payloads;
+ this.retries = retries;
}
public String getServiceName() {
@@ -85,6 +87,14 @@ public class AddTxEventDomain implements DomainEvent {
this.compensationMethod = compensationMethod;
}
+ public int getRetries() {
+ return retries;
+ }
+
+ public void setRetries(int retries) {
+ this.retries = retries;
+ }
+
public byte[] getPayloads() {
return payloads;
}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/consumer/SagaEventConsumer.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/consumer/SagaEventConsumer.java
index fa481f8..01a7327 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/consumer/SagaEventConsumer.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/consumer/SagaEventConsumer.java
@@ -25,6 +25,8 @@ import akka.util.Timeout;
import com.google.common.base.Optional;
import com.google.common.eventbus.Subscribe;
import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.servicecomb.pack.alpha.fsm.SagaActor;
import org.apache.servicecomb.pack.alpha.fsm.event.base.BaseEvent;
@@ -42,21 +44,31 @@ public class SagaEventConsumer {
@Autowired
ActorSystem system;
+ private Map<String,ActorRef> sagaCache = new HashMap<>();
+
/**
* Receive fsm message
* */
@Subscribe
public void receiveSagaEvent(BaseEvent event) throws Exception {
- LOG.info("receive {} ", event.toString());
+ if(LOG.isDebugEnabled()){
+ LOG.debug("receive {} ", event.toString());
+ }
try{
ActorRef saga;
- String actorPath = "/user/" + event.getGlobalTxId();
- Optional<ActorRef> optional = this.getActorRefFromPath(actorPath);
- if (!optional.isPresent()) {
+ if(sagaCache.containsKey(event.getGlobalTxId())){
+ saga = sagaCache.get(event.getGlobalTxId());
+ }else{
saga = system.actorOf(SagaActor.props(event.getGlobalTxId()), event.getGlobalTxId());
- } else {
- saga = optional.get();
+ sagaCache.put(event.getGlobalTxId(), saga);
}
+// String actorPath = "/user/" + event.getGlobalTxId();
+// Optional<ActorRef> optional = this.getActorRefFromPath(actorPath);
+// if (!optional.isPresent()) {
+// saga = system.actorOf(SagaActor.props(event.getGlobalTxId()), event.getGlobalTxId());
+// } else {
+// saga = optional.get();
+// }
saga.tell(event, ActorRef.noSender());
if(LOG.isDebugEnabled()){
LOG.debug("tell {} to {}", event.toString(),saga);
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 de849e2..ce0fb5b 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
@@ -18,6 +18,7 @@
package org.apache.servicecomb.pack.alpha.fsm.model;
import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.servicecomb.pack.alpha.fsm.TxState;
public class TxEntity implements Serializable {
@@ -32,6 +33,8 @@ public class TxEntity implements Serializable {
private String compensationMethod;
private byte[] payloads;
private byte[] throwablePayLoads;
+ private int retries;
+ private AtomicInteger retriesCounter = new AtomicInteger();
public String getServiceName() {
return serviceName;
@@ -121,6 +124,18 @@ public class TxEntity implements Serializable {
this.throwablePayLoads = throwablePayLoads;
}
+ public int getRetries() {
+ return retries;
+ }
+
+ public void setRetries(int retries) {
+ this.retries = retries;
+ }
+
+ public AtomicInteger getRetriesCounter() {
+ return retriesCounter;
+ }
+
public static Builder builder() {
return new Builder();
}
@@ -188,6 +203,11 @@ public class TxEntity implements Serializable {
return this;
}
+ public Builder retries(int retries) {
+ txEntity.setRetries(retries);
+ return this;
+ }
+
public TxEntity build() {
return txEntity;
}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SpringAkkaExtension.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SpringAkkaExtension.java
index 6994084..6346d80 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SpringAkkaExtension.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SpringAkkaExtension.java
@@ -17,7 +17,7 @@
package org.apache.servicecomb.pack.alpha.fsm.spring.integration.akka;
-import static org.apache.servicecomb.pack.common.EventType.TxStartedEvent;
+import static org.apache.servicecomb.pack.common.EventType.TxCompensateEvent;
import akka.actor.AbstractExtensionId;
import akka.actor.ExtendedActorSystem;
@@ -57,11 +57,10 @@ public class SpringAkkaExtension extends AbstractExtensionId<SpringExt> {
txEntity.getGlobalTxId(),
txEntity.getLocalTxId(),
txEntity.getParentTxId(),
- TxStartedEvent.name(),
+ TxCompensateEvent.name(),
txEntity.getCompensationMethod(),
txEntity.getPayloads());
omegaCallback.compensate(event);
- LOG.info(omegaCallback.toString());
} else {
LOG.warn("Spring Bean {} doesn't exist in ApplicationContext", omegaCallbackBeanName);
}
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 3bbaf6d..fc9f3d3 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
@@ -405,4 +405,71 @@ 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 compensateOmegaDisconnectTest() {
+ 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();
+ final Map<String, OmegaCallback>[] omegaInstance = new Map[]{null};
+ final String[] serviceName = new String[1];
+ omegaEventSender.getOmegaEventSagaSimulator().lastTxAbortedEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
+ if(event.getType().equals(EventType.TxAbortedEvent.name())){
+ //simulate omega disconnect
+ serviceName[0] = event.getServiceName();
+ omegaInstance[0] = omegaEventSender.getOmegaCallbacks().get(event.getServiceName());
+ omegaEventSender.getOmegaCallbacks().remove(event.getServiceName());
+ }
+ omegaEventSender.getBlockingStub().onTxEvent(event);
+ });
+ //simulate omega connected
+ await().atMost(2, SECONDS).until(() -> {
+ SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId);
+ return sagaData !=null && sagaData.getTxEntityMap().size()==3;
+ });
+ omegaEventSender.getOmegaCallbacks().put(serviceName[0], omegaInstance[0]);
+
+ await().atMost(2, 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);
+ Assert.assertArrayEquals(sagaData.getTxEntityMap().get(localTxId_3).getThrowablePayLoads(),NullPointerException.class.getName().getBytes());
+ }
+
+ @Test
+ public void compensateFailRetryOnce() {
+ 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().lastTxAbortedEvents(globalTxId, localTxId_1, localTxId_2, localTxId_3).stream().forEach( event -> {
+ omegaEventSender.getBlockingStub().onTxEvent(event);
+ });
+ await().atMost(2, 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);
+ Assert.assertArrayEquals(sagaData.getTxEntityMap().get(localTxId_3).getThrowablePayLoads(),NullPointerException.class.getName().getBytes());
+ }
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSender.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSender.java
index a6a5c3c..fe1f965 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSender.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSender.java
@@ -85,8 +85,8 @@ public class OmegaEventSender {
return blockingStub;
}
- public Map<String, OmegaCallback> getOmegaCallbacks(){
- return omegaCallbacks.get(serviceName);
+ public Map<String, Map<String, OmegaCallback>> getOmegaCallbacks(){
+ return omegaCallbacks;
}
public void reset(){