[Fix #1395] Refining approach

There were some implicit constraints in the map to be returned by
retrieveEvents.
I think it is better to pass the Map already populated with the expected
registrations associated to an empty modifiable array and let the
implementor just add the cloud events to every array.

Also, the event can be stored at the end, reducing the likeness of using
it for calculations in a different cluster

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

impl/core/src/main/java/io/serverlessworkflow/impl/scheduler/ScheduledEventConsumer.java

@@ -60,11 +60,10 @@ public ScheduledEventConsumer(
           builderInfo.registrations().registrations();
       allStrategyCorrelationInfo.init(registrationBuilders, this::start);
       registrationBuilders.forEach(
-          reg -> {
-            registrations.add(
-                eventConsumer.register(
-                    reg, ce -> allStrategyCorrelationInfo.correlate(reg, (CloudEvent) ce)));
-          });
+          reg ->
+              registrations.add(
+                  eventConsumer.register(
+                      reg, ce -> allStrategyCorrelationInfo.correlate(reg, (CloudEvent) ce))));
     } else {
       builderInfo
           .registrations()

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/AbstractAllStrategyCorrelationInfo.java

@@ -90,19 +90,33 @@ private Collection<Map<EventRegistrationBuilder, CloudEvent>> eventAdded(
       CorrelationOperations operations, String reg, CloudEvent event) {
     logger.debug(
         "Received event {} for definition {} and registration {}", event, definition.id(), reg);
+    Map<String, List<CloudEvent>> events = initMap();
+    operations.retrieveEvents(events);
+    events.get(reg).add(event);
+    Collection<Map<EventRegistrationBuilder, CloudEvent>> result = checkCorrelation(events);
     operations.storeEvent(reg, event);
-    return checkCorrelation(operations);
+    markProcessed(operations, result);
+    return result;
+  }
+
+  private Map<String, List<CloudEvent>> initMap() {
+    return id2RegMapping.keySet().stream()
+        .collect(Collectors.toMap(k -> k, k -> new ArrayList<>()));
   }
 
   private Collection<Map<EventRegistrationBuilder, CloudEvent>> startupCheck(
       CorrelationOperations operations) {
+    logger.debug("Checking cloud events for definition {}", definition.id());
     operations.clearProcessed();
-    return checkCorrelation(operations);
+    Map<String, List<CloudEvent>> events = initMap();
+    operations.retrieveEvents(events);
+    Collection<Map<EventRegistrationBuilder, CloudEvent>> result = checkCorrelation(events);
+    markProcessed(operations, result);
+    return result;
   }
 
   private final Collection<Map<EventRegistrationBuilder, CloudEvent>> checkCorrelation(
-      CorrelationOperations operations) {
-    Map<String, List<CloudEvent>> events = operations.retrieveEvents(id2RegMapping.keySet());
+      Map<String, List<CloudEvent>> events) {
     logger.debug("Stored CloudEvents for definition {} are {}", definition.id(), events);
     if (events.isEmpty()) {
       return List.of();
@@ -117,13 +131,18 @@ private final Collection<Map<EventRegistrationBuilder, CloudEvent>> checkCorrela
           notDone = false;
           break;
         }
-        CloudEvent retrieved = list.remove(0);
-        row.put(id2RegMapping.get(item.getKey()), retrieved);
+        row.put(id2RegMapping.get(item.getKey()), list.remove(0));
       }
       if (notDone) {
         result.add(row);
       }
     }
+    return result;
+  }
+
+  private void markProcessed(
+      CorrelationOperations operations,
+      Collection<Map<EventRegistrationBuilder, CloudEvent>> result) {
     if (!result.isEmpty()) {
       Map<String, Collection<String>> processed = new HashMap<>();
       for (Map<EventRegistrationBuilder, CloudEvent> item : result) {
@@ -135,7 +154,6 @@ private final Collection<Map<EventRegistrationBuilder, CloudEvent>> checkCorrela
       }
       operations.markAsProcessed(processed);
     }
-    return result;
   }
 
   public void addMetadata(

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/CorrelationOperations.java

@@ -22,9 +22,7 @@
 
 interface CorrelationOperations {
 
-  default Map<String, List<CloudEvent>> retrieveEvents(Collection<String> targetRegIds) {
-    return Map.of();
-  }
+  default void retrieveEvents(Map<String, List<CloudEvent>> reg2EventsMap) {}
 
   default void storeEvent(String regId, CloudEvent event) {}
 

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BigMapInstanceTransaction.java

@@ -29,7 +29,6 @@
 import io.serverlessworkflow.impl.persistence.PersistenceWorkflowInfo;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -113,26 +112,29 @@ public void clearStatus(WorkflowContextData workflowContext) {
     clearStatus(workflowContext.definition(), key(workflowContext));
   }
 
-  public Map<String, List<CloudEvent>> retrieveEvents(Collection<String> targetRegIds) {
-    Map<String, List<CloudEvent>> result = new HashMap<>();
-    targetRegIds.forEach(
-        regId -> {
-          Map<String, P> processedCes = processedCloudEvents(regId);
-          Map<String, C> ces = cloudEvents(regId);
-          result.put(
-              regId,
-              ces.values().stream()
-                  .map(this::unmarshallCloudEvent)
-                  .filter(ce -> !processedCes.containsKey(ce.getId()))
-                  .collect(Collectors.toCollection(ArrayList::new)));
-        });
-    return result;
+  @Override
+  public void retrieveEvents(Map<String, List<CloudEvent>> events) {
+    events
+        .entrySet()
+        .forEach(
+            e -> {
+              String regId = e.getKey();
+              Map<String, P> processedCes = processedCloudEvents(regId);
+              Map<String, C> ces = cloudEvents(regId);
+              e.setValue(
+                  ces.values().stream()
+                      .map(this::unmarshallCloudEvent)
+                      .filter(ce -> !processedCes.containsKey(ce.getId()))
+                      .collect(Collectors.toCollection(ArrayList::new)));
+            });
   }
 
+  @Override
   public void storeEvent(String regId, CloudEvent event) {
     cloudEvents(regId).put(event.getId(), marshallCloudEvent(event));
   }
 
+  @Override
   public void markAsProcessed(Map<String, Collection<String>> regCeIds) {
     regCeIds.forEach(
         (k, v) -> {
@@ -141,6 +143,7 @@ public void markAsProcessed(Map<String, Collection<String>> regCeIds) {
         });
   }
 
+  @Override
   public void clearProcessed() {
     deleteAllProcessedMaps();
   }