[Fix #1395] Handling duplicated id

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

Author: fjtirado

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,43 @@ 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)