SCB-547 try to fix the EventScanner multiple AbortedGlobalTx issue (#186

)
apache · May 8, 2018 · ef3d7b9 · ef3d7b9
1 parent 9cb8f22
commit ef3d7b9
Show file tree

Hide file tree

Showing 5 changed files with 14 additions and 7 deletions.
diff --git a/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/EventScanner.java b/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/EventScanner.java
@@ -25,6 +25,7 @@
 import static org.apache.servicecomb.saga.common.EventType.TxStartedEvent;
 
 import java.lang.invoke.MethodHandles;
+import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 
 import org.slf4j.Logger;
@@ -141,20 +142,26 @@ private void abortTimeoutEvents() {
   }
 
   private void updateTransactionStatus() {
-    eventRepository.findFirstAbortedGlobalTransaction().ifPresent(this::markGlobalTxEnd);
+    eventRepository.findFirstAbortedGlobalTransaction().ifPresent(this::markGlobalTxEndWithEvents);
   }
 
   private void markSagaEnded(TxEvent event) {
     if (commandRepository.findUncompletedCommands(event.globalTxId()).isEmpty()) {
-      markGlobalTxEnd(event);
+      markGlobalTxEntWithEvent(event);
     }
   }
 
-  private void markGlobalTxEnd(TxEvent event) {
+  private void markGlobalTxEntWithEvent(TxEvent event) {
     eventRepository.save(toSagaEndedEvent(event));
     LOG.info("Marked end of transaction with globalTxId {}", event.globalTxId());
   }
 
+  private void markGlobalTxEndWithEvents(List<TxEvent> events) {
+    events.forEach(event -> {
+      markGlobalTxEntWithEvent(event);
+    });
+  }
+
   private TxEvent toTxAbortedEvent(TxTimeout timeout) {
     return new TxEvent(
         timeout.serviceName(),

diff --git a/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/TxEventRepository.java b/alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/TxEventRepository.java
@@ -42,7 +42,7 @@ public interface TxEventRepository {
    * </ol>
    * @return
    */
-  Optional<TxEvent> findFirstAbortedGlobalTransaction();
+  Optional<List<TxEvent>> findFirstAbortedGlobalTransaction();
 
   /**
    * Find timeout {@link TxEvent}s. A timeout TxEvent satisfies below requirements:

diff --git a/...ha-core/src/test/java/org/apache/servicecomb/saga/alpha/core/TxConsistentServiceTest.java b/...ha-core/src/test/java/org/apache/servicecomb/saga/alpha/core/TxConsistentServiceTest.java
@@ -49,7 +49,7 @@ public void save(TxEvent event) {
     }
 
     @Override
-    public Optional<TxEvent> findFirstAbortedGlobalTransaction() {
+    public Optional<List<TxEvent>> findFirstAbortedGlobalTransaction() {
       return Optional.empty();
     }
 

diff --git a/...erver/src/main/java/org/apache/servicecomb/saga/alpha/server/SpringTxEventRepository.java b/...erver/src/main/java/org/apache/servicecomb/saga/alpha/server/SpringTxEventRepository.java
@@ -40,7 +40,7 @@ public void save(TxEvent event) {
   }
 
   @Override
-  public Optional<TxEvent> findFirstAbortedGlobalTransaction() {
+  public Optional<List<TxEvent>> findFirstAbortedGlobalTransaction() {
     return eventRepo.findFirstAbortedGlobalTxByType();
   }
 

diff --git a/...ver/src/main/java/org/apache/servicecomb/saga/alpha/server/TxEventEnvelopeRepository.java b/...ver/src/main/java/org/apache/servicecomb/saga/alpha/server/TxEventEnvelopeRepository.java
@@ -46,7 +46,7 @@ interface TxEventEnvelopeRepository extends CrudRepository<TxEvent, Long> {
       + "  AND t2.localTxId = t.localTxId "
       + "  AND t2.type = 'TxStartedEvent') = 0 "
       + "OR t.globalTxId = t.localTxId)")
-  Optional<TxEvent> findFirstAbortedGlobalTxByType();
+  Optional<List<TxEvent>> findFirstAbortedGlobalTxByType();
 
   @Query("SELECT t FROM TxEvent t "
       + "WHERE t.type IN ('TxStartedEvent', 'SagaStartedEvent') "