#2878: fixed setting FSP on non-transactional projections

factcast · Jun 13, 2024 · 144912e · 144912e
1 parent cef7175
commit 144912e
Show file tree

Hide file tree

Showing 2 changed files with 78 additions and 4 deletions.
diff --git a/factcast-factus/src/main/java/org/factcast/factus/projector/ProjectorImpl.java b/factcast-factus/src/main/java/org/factcast/factus/projector/ProjectorImpl.java
@@ -98,6 +98,7 @@ public void doApply(@NonNull List<Fact> facts) {
       try {
         callHandlerFor(f);
         latestSuccessful = FactStreamPosition.from(f);
+        setFactStreamPositionIfAwareButNotTransactional(latestSuccessful);
       } catch (Exception e) {
         log.trace(
             "returned with Exception {}:",
@@ -113,10 +114,11 @@ public void doApply(@NonNull List<Fact> facts) {
     } // end loop
 
     try {
-      // this is something we only do, if the whole batch was successfully applied
-      if (latestSuccessful != null) {
-        setFactStreamPositionIfAware(latestSuccessful);
-      }
+      if (projection instanceof TransactionAware)
+        // this is something we only do, if the whole batch was successfully applied
+        if (latestSuccessful != null) {
+          setFactStreamPositionIfAware(latestSuccessful);
+        }
     } catch (Exception e) {
 
       rollbackIfTransactional();
@@ -135,6 +137,14 @@ public void doApply(@NonNull List<Fact> facts) {
     }
   }
 
+  private void setFactStreamPositionIfAwareButNotTransactional(
+      @NonNull FactStreamPosition latestSuccessful) {
+    if (!(projection instanceof TransactionAware)
+        && projection instanceof FactStreamPositionAware) {
+      ((FactStreamPositionAware) projection).factStreamPosition(latestSuccessful);
+    }
+  }
+
   @VisibleForTesting
   void retryApplicableIfTransactional(List<Fact> facts, Fact f) {
     if (projection instanceof TransactionAware) {

diff --git a/factcast-factus/src/test/java/org/factcast/factus/projector/ProjectorImplTest.java b/factcast-factus/src/test/java/org/factcast/factus/projector/ProjectorImplTest.java
@@ -20,6 +20,7 @@
 
 import com.google.common.collect.Lists;
 import java.lang.reflect.Method;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
@@ -38,6 +39,7 @@
 import org.factcast.factus.*;
 import org.factcast.factus.event.DefaultEventSerializer;
 import org.factcast.factus.event.EventSerializer;
+import org.factcast.factus.projection.FactStreamPositionAware;
 import org.factcast.factus.projection.Projection;
 import org.factcast.factus.projection.parameter.HandlerParameterContributors;
 import org.factcast.factus.projection.tx.OpenTransactionAware;
@@ -47,6 +49,7 @@
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
+@SuppressWarnings("deprecation")
 class ProjectorImplTest {
 
   private final DefaultEventSerializer eventSerializer =
@@ -699,4 +702,65 @@ void determinesTypeParameter() {
           .isSameAs(SomeTransactionInterface.class);
     }
   }
+
+  @Nested
+  class WhenApplyingToFSPAwareNonTx {
+
+    class FSAProjection implements FactStreamPositionAware, Projection {
+      private FactStreamPosition p;
+
+      @Nullable
+      @Override
+      public FactStreamPosition factStreamPosition() {
+        return p;
+      }
+
+      @Override
+      public void factStreamPosition(@NonNull FactStreamPosition factStreamPosition) {
+        this.p = factStreamPosition;
+      }
+
+      @HandlerFor(ns = "test", type = "test")
+      void apply(Fact f) {
+        if (f.header().meta().containsKey("pleaseThrow"))
+          throw new RuntimeException("you asked me to");
+      }
+    }
+
+    @Test
+    void hasFSPosSetToFactBeforeErrorOccured() {
+      Fact f1 = Fact.builder().ns("test").type("test").build("");
+      Fact f2 = Fact.builder().ns("test").type("test").build("");
+      Fact f3 = Fact.builder().ns("test").type("test").meta("pleaseThrow", "").build("");
+      Fact f4 = Fact.builder().ns("test").type("test").build("");
+
+      ArrayList<Fact> facts = Lists.newArrayList(f1, f2, f3, f4);
+
+      FSAProjection projection = new FSAProjection();
+      ProjectorImpl<Projection> uut = new ProjectorImpl<>(projection, eventSerializer);
+      assertThatThrownBy(
+              () -> {
+                uut.apply(facts);
+              })
+          .isInstanceOf(Exception.class);
+
+      Assertions.assertThat(projection.factStreamPosition()).isEqualTo(FactStreamPosition.from(f2));
+    }
+
+    @Test
+    void hasFSPosSetOnEveryApply() {
+      Fact f1 = Fact.builder().ns("test").type("test").build("");
+      Fact f2 = Fact.builder().ns("test").type("test").build("");
+      Fact f3 = Fact.builder().ns("test").type("test").build("");
+      Fact f4 = Fact.builder().ns("test").type("test").build("");
+
+      ArrayList<Fact> facts = Lists.newArrayList(f1, f2, f3, f4);
+
+      FSAProjection projection = spy(new FSAProjection());
+      ProjectorImpl<Projection> uut = new ProjectorImpl<>(projection, eventSerializer);
+      uut.apply(facts);
+
+      verify(projection, times(4)).factStreamPosition(any());
+    }
+  }
 }