[Fix #1395] Handling duplicated id

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

Author: fjtirado

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

@@ -17,10 +17,10 @@
 
 import io.cloudevents.CloudEvent;
 import io.serverlessworkflow.impl.events.EventRegistrationBuilder;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.function.Consumer;
 
@@ -37,7 +37,7 @@ public static AllStrategyCorrelationInfo instance() {
 
   private InMemoryAllStrategyCorrelationInfo() {}
 
-  private Map<EventRegistrationBuilder, List<CloudEvent>> correlatedEvents;
+  private Map<EventRegistrationBuilder, Collection<CloudEvent>> correlatedEvents;
   private Consumer<Map<EventRegistrationBuilder, CloudEvent>> starter;
 
   @Override
@@ -48,9 +48,11 @@ public void correlate(EventRegistrationBuilder reg, CloudEvent event) {
     synchronized (correlatedEvents) {
       correlatedEvents.get(reg).add(event);
       if (satisfyCondition(correlatedEvents)) {
-        for (java.util.Map.Entry<EventRegistrationBuilder, List<CloudEvent>> values :
+        for (java.util.Map.Entry<EventRegistrationBuilder, Collection<CloudEvent>> values :
             correlatedEvents.entrySet()) {
-          result.put(values.getKey(), values.getValue().remove(0));
+          Iterator<CloudEvent> iter = values.getValue().iterator();
+          result.put(values.getKey(), iter.next());
+          iter.remove();
         }
       }
     }
@@ -65,11 +67,11 @@ public void init(
       Consumer<Map<EventRegistrationBuilder, CloudEvent>> starter) {
     correlatedEvents = new HashMap<>();
     this.starter = starter;
-    regs.forEach(reg -> correlatedEvents.put(reg, new ArrayList<CloudEvent>()));
+    regs.forEach(reg -> correlatedEvents.put(reg, new LinkedHashSet<CloudEvent>()));
   }
 
-  private boolean satisfyCondition(Map<EventRegistrationBuilder, List<CloudEvent>> events) {
-    for (List<CloudEvent> values : events.values()) {
+  private boolean satisfyCondition(Map<EventRegistrationBuilder, Collection<CloudEvent>> events) {
+    for (Collection<CloudEvent> values : events.values()) {
       if (values.isEmpty()) {
         return false;
       }

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

@@ -23,6 +23,8 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -90,7 +92,7 @@ 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();
+    Map<String, Collection<CloudEvent>> events = initMap();
     operations.retrieveEvents(events);
     events.get(reg).add(event);
     Collection<Map<EventRegistrationBuilder, CloudEvent>> result = checkCorrelation(events);
@@ -99,39 +101,40 @@ private Collection<Map<EventRegistrationBuilder, CloudEvent>> eventAdded(
     return result;
   }
 
-  private Map<String, List<CloudEvent>> initMap() {
+  private Map<String, Collection<CloudEvent>> initMap() {
     return id2RegMapping.keySet().stream()
-        .collect(Collectors.toMap(k -> k, k -> new ArrayList<>()));
+        .collect(Collectors.toMap(k -> k, k -> new LinkedHashSet<>()));
   }
 
   private Collection<Map<EventRegistrationBuilder, CloudEvent>> startupCheck(
       CorrelationOperations operations) {
     logger.debug("Checking cloud events for definition {}", definition.id());
     operations.clearProcessed();
-    Map<String, List<CloudEvent>> events = initMap();
+    Map<String, Collection<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(
-      Map<String, List<CloudEvent>> events) {
+      Map<String, Collection<CloudEvent>> events) {
     logger.debug("Stored CloudEvents for definition {} are {}", definition.id(), events);
-    if (events.isEmpty()) {
-      return List.of();
-    }
     Collection<Map<EventRegistrationBuilder, CloudEvent>> result = new ArrayList<>();
+    Map<String, Iterator<CloudEvent>> iteratingEvents =
+        events.entrySet().stream()
+            .collect(Collectors.toMap(Entry::getKey, e -> e.getValue().iterator()));
     boolean notDone = true;
     while (notDone) {
       Map<EventRegistrationBuilder, CloudEvent> row = new HashMap<>();
-      for (Entry<String, List<CloudEvent>> item : events.entrySet()) {
-        List<CloudEvent> list = item.getValue();
-        if (list.isEmpty()) {
+      for (Entry<String, Iterator<CloudEvent>> item : iteratingEvents.entrySet()) {
+        Iterator<CloudEvent> iter = item.getValue();
+        if (!iter.hasNext()) {
           notDone = false;
           break;
         }
-        row.put(id2RegMapping.get(item.getKey()), list.remove(0));
+        row.put(id2RegMapping.get(item.getKey()), iter.next());
+        iter.remove();
       }
       if (notDone) {
         result.add(row);

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

@@ -17,12 +17,11 @@
 
 import io.cloudevents.CloudEvent;
 import java.util.Collection;
-import java.util.List;
 import java.util.Map;
 
 interface CorrelationOperations {
 
-  default void retrieveEvents(Map<String, List<CloudEvent>> reg2EventsMap) {}
+  default void retrieveEvents(Map<String, Collection<CloudEvent>> reg2EventsMap) {}
 
   default void storeEvent(String regId, CloudEvent event) {}
 

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

@@ -28,7 +28,6 @@
 import io.serverlessworkflow.impl.persistence.PersistenceTaskInfo;
 import io.serverlessworkflow.impl.persistence.PersistenceWorkflowInfo;
 import java.util.Collection;
-import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
@@ -112,13 +111,13 @@ public void clearStatus(WorkflowContextData workflowContext) {
   }
 
   @Override
-  public void retrieveEvents(Map<String, List<CloudEvent>> events) {
+  public void retrieveEvents(Map<String, Collection<CloudEvent>> events) {
     events
         .entrySet()
         .forEach(
             e -> {
               String regId = e.getKey();
-              List<CloudEvent> cloudEvents = e.getValue();
+              Collection<CloudEvent> cloudEvents = e.getValue();
               Map<String, P> processedCes = processedCloudEvents(regId);
               cloudEvents(regId).values().stream()
                   .map(this::unmarshallCloudEvent)