This is an automated email from the ASF dual-hosted git repository. ningjiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit b54421e855f2590a646472de15943481788a017e Author: Lei Zhang <[email protected]> AuthorDate: Fri Jul 26 19:08:00 2019 +0800 SCB-1369 Clean up memory data after persisting data to ES --- .../spring/integration/akka/SagaDataExtension.java | 63 +++---------------- .../servicecomb/pack/alpha/fsm/SagaActorTest.java | 33 +++++----- .../pack/alpha/fsm/SagaIntegrationTest.java | 73 +++++++++++----------- 3 files changed, 64 insertions(+), 105 deletions(-) diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SagaDataExtension.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SagaDataExtension.java index 5354f21..dbf7425 100644 --- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SagaDataExtension.java +++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/spring/integration/akka/SagaDataExtension.java @@ -39,8 +39,6 @@ public class SagaDataExtension extends AbstractExtensionId<SagaDataExt> { private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); public static final SagaDataExtension SAGA_DATA_EXTENSION_PROVIDER = new SagaDataExtension(); - //TODO We could use test profile the enable this kind feature - public static boolean autoCleanSagaDataMap = true; // Only for Test @Override public SagaDataExt createExtension(ExtendedActorSystem system) { @@ -48,30 +46,18 @@ public class SagaDataExtension extends AbstractExtensionId<SagaDataExt> { } public static class SagaDataExt implements Extension { - + private SagaData lastSagaData; private final ConcurrentHashMap<String, SagaData> sagaDataMap = new ConcurrentHashMap(); - private String lastGlobalTxId; - private CleanMemForTest cleanMemForTest = new CleanMemForTest(sagaDataMap); private MetricsService metricsService; private TransactionRepositoryChannel repositoryChannel; - public SagaDataExt() { - // Just to avoid the overflow of the OldGen for stress testing - // Delete after SagaData persistence - if (autoCleanSagaDataMap) { - new Thread(cleanMemForTest).start(); - } - } - public void putSagaData(String globalTxId, SagaData sagaData) { - lastGlobalTxId = globalTxId; sagaDataMap.put(globalTxId, sagaData); + lastSagaData = sagaData; } public void stopSagaData(String globalTxId, SagaData sagaData) { - // TODO save SagaDate to database and clean sagaDataMap this.putSagaData(globalTxId, sagaData); - lastGlobalTxId = globalTxId; if (sagaData.getLastState() == SagaActorState.COMMITTED) { this.metricsService.metrics().doCommitted(); } else if (sagaData.getLastState() == SagaActorState.COMPENSATED) { @@ -102,22 +88,17 @@ public class SagaDataExtension extends AbstractExtensionId<SagaDataExt> { .events(sagaData.getEvents()) .build(); repositoryChannel.send(record); + sagaDataMap.remove(globalTxId); } - public SagaData getSagaData(String globalTxId) { - // TODO If globalTxId does not exist in sagaDataMap then - // load from the database - return sagaDataMap.get(globalTxId); + // Only for Test + public SagaData getLastSagaDate() { + return lastSagaData; } - // Only test - public void clearSagaData() { - lastGlobalTxId = null; - sagaDataMap.clear(); - } - - public SagaData getLastSagaData() { - return getSagaData(lastGlobalTxId); + // Only for Test + public void cleanLastSagaData() { + lastSagaData = null; } public void doSagaBeginCounter() { @@ -142,30 +123,4 @@ public class SagaDataExtension extends AbstractExtensionId<SagaDataExt> { this.repositoryChannel = repositoryChannel; } } - - static class CleanMemForTest implements Runnable { - - final ConcurrentHashMap<String, SagaData> sagaDataMap; - - public CleanMemForTest(ConcurrentHashMap<String, SagaData> sagaDataMap) { - this.sagaDataMap = sagaDataMap; - } - - @Override - public void run() { - while (true) { - try { - sagaDataMap.clear(); - } catch (Exception e) { - LOG.error(e.getMessage(), e); - } finally { - try { - Thread.sleep(10000); - } catch (InterruptedException e) { - LOG.error(e.getMessage(), e); - } - } - } - } - } } diff --git a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java index 198ebf4..94e30a3 100644 --- a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java +++ b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java @@ -41,6 +41,7 @@ import org.apache.servicecomb.pack.alpha.fsm.repository.channel.MemoryTransactio import org.apache.servicecomb.pack.alpha.fsm.repository.elasticsearch.ElasticsearchTransactionRepository; import org.apache.servicecomb.pack.alpha.fsm.repository.TransactionRepository; import org.apache.servicecomb.pack.alpha.fsm.spring.integration.akka.SagaDataExtension; +import org.junit.After; import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; @@ -85,7 +86,6 @@ public class SagaActorTest { @BeforeClass public static void setup() { - SagaDataExtension.autoCleanSagaDataMap=false; system = ActorSystem.create("SagaActorTest", ConfigFactory.parseMap(getPersistenceMemConfig())); SAGA_DATA_EXTENSION_PROVIDER.get(system).setMetricsService(metricsService); } @@ -104,6 +104,11 @@ public class SagaActorTest { SAGA_DATA_EXTENSION_PROVIDER.get(system).setRepositoryChannel(repositoryChannel); } + @After + public void after(){ + SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).cleanLastSagaData(); + } + public String genPersistenceId() { return UUID.randomUUID().toString(); } @@ -166,7 +171,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); sagaData.getTxEntityMap().forEach((k, v) -> { @@ -254,7 +259,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), recoveredSaga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); sagaData.getTxEntityMap().forEach((k, v) -> { @@ -306,7 +311,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 1); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.FAILED); @@ -368,7 +373,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 2); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED); @@ -441,7 +446,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED); @@ -514,7 +519,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED); @@ -576,7 +581,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.FAILED); @@ -654,7 +659,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED); @@ -727,7 +732,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMMITTED); @@ -796,7 +801,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertThat(eventList, is(sagaData.getEvents())); system.stop(saga); @@ -850,7 +855,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); sagaData.getTxEntityMap().forEach((k, v) -> { @@ -913,7 +918,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); sagaData.getTxEntityMap().forEach((k, v) -> { @@ -975,7 +980,7 @@ public class SagaActorTest { Terminated terminated = expectMsgClass(Terminated.class); assertEquals(terminated.getActor(), saga); - SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertEquals(sagaData.getGlobalTxId(), globalTxId); assertEquals(sagaData.getTxEntityMap().size(), 3); assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED); 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 c696031..1cff92c 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 @@ -25,15 +25,15 @@ import static org.junit.Assert.assertNotNull; import akka.actor.ActorSystem; import java.util.UUID; import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService; -import org.apache.servicecomb.pack.alpha.fsm.sink.SagaActorEventSender; import org.apache.servicecomb.pack.alpha.fsm.model.SagaData; +import org.apache.servicecomb.pack.alpha.fsm.sink.SagaActorEventSender; import org.apache.servicecomb.pack.alpha.fsm.spring.integration.akka.SagaDataExtension; -import org.junit.BeforeClass; +import org.junit.After; import org.junit.Test; import org.junit.runner.RunWith; -import org.mockito.Mock; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.mock.mockito.MockBean; import org.springframework.data.elasticsearch.core.ElasticsearchTemplate; import org.springframework.test.context.junit4.SpringRunner; @@ -42,17 +42,16 @@ import org.springframework.test.context.junit4.SpringRunner; properties = { //akka "alpha.feature.akka.enabled=true", - "alpha.feature.akka.channel.type=memory", "akkaConfig.akka.persistence.journal.plugin=akka.persistence.journal.inmem", "akkaConfig.akka.persistence.journal.leveldb.dir=target/example/journal", "akkaConfig.akka.persistence.snapshot-store.plugin=akka.persistence.snapshot-store.local", "akkaConfig.akka.persistence.snapshot-store.local.dir=target/example/snapshots", //elasticsearch - "alpha.feature.akka.transcation.repository.channel.type=memory", - "alpha.feature.akka.transcation.repository.type=elasticsearch", - "spring.data.elasticsearch.cluster-name=alpha-cluster", - "spring.data.elasticsearch.cluster-nodes=localhost:9300", - "spring.elasticsearch.rest.uris=http://localhost:9200"; + "alpha.feature.akka.transaction.repository.channel.type=memory", + "alpha.feature.akka.transaction.repository.type=elasticsearch", + "alpha.feature.akka.transaction.repository.elasticsearch.batchSize=0", + "spring.data.elasticsearch.cluster-name=elasticsearch", + "spring.data.elasticsearch.cluster-nodes=localhost:9300" }) public class SagaIntegrationTest { @@ -65,12 +64,12 @@ public class SagaIntegrationTest { @Autowired MetricsService metricsService; - @Mock + @MockBean ElasticsearchTemplate elasticsearchTemplate; - @BeforeClass - public static void setup(){ - SagaDataExtension.autoCleanSagaDataMap=false; + @After + public void after(){ + SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).cleanLastSagaData(); } @Test @@ -83,10 +82,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMMITTED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -108,10 +107,10 @@ public class SagaIntegrationTest { }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMPENSATED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),1); @@ -127,10 +126,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMPENSATED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),2); @@ -148,10 +147,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMPENSATED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -170,10 +169,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMPENSATED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -192,10 +191,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMPENSATED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -214,10 +213,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMPENSATED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -236,10 +235,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.SUSPENDED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -259,10 +258,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(timeout + 2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.SUSPENDED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -281,10 +280,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMMITTED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -303,10 +302,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMMITTED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3); @@ -325,10 +324,10 @@ public class SagaIntegrationTest { sagaActorEventSender.send(event); }); await().atMost(2, SECONDS).until(() -> { - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); return sagaData !=null && sagaData.isTerminated() && sagaData.getLastState()==SagaActorState.COMPENSATED; }); - SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getSagaData(globalTxId); + SagaData sagaData = SagaDataExtension.SAGA_DATA_EXTENSION_PROVIDER.get(system).getLastSagaDate(); assertNotNull(sagaData.getBeginTime()); assertNotNull(sagaData.getEndTime()); assertEquals(sagaData.getTxEntityMap().size(),3);
