[Fix #1354] Adding test for fake failure reported by IA (#1360)

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

Author: fjtirado

experimental/test/src/test/java/io/serverlessworkflow/fluent/test/ForEachFuncTest.java

@@ -17,12 +17,20 @@
 
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.*;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.data.PojoCloudEventData;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.events.InMemoryEvents;
+import java.time.Duration;
+import java.util.Collection;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
 import org.junit.jupiter.api.Test;
 
 public class ForEachFuncTest {
@@ -33,8 +41,10 @@ private static record EnhancedOrder(String id, int salary) {}
 
   private static record OrdersPayload(List<Order> orders) {}
 
+  private static record OrderName(String id, String name) {}
+
   @Test
-  void testForEachIteration() throws Exception {
+  void testForEachIteration() {
 
     Workflow workflow =
         FuncWorkflowBuilder.workflow("foreach-workflow")
@@ -51,6 +61,55 @@ void testForEachIteration() throws Exception {
     }
   }
 
+  @Test
+  void testForEachEmit() {
+
+    String eventType = "test.item.emitted";
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("forEach-bug-reproducer")
+            .tasks(
+                // ForEach should emit 3 events, one per item
+                forEach(
+                    (Collection<Map<String, String>> items) -> items,
+                    emitJson(eventType, Map.class).inputFrom("$item")))
+            .build();
+
+    List<CloudEvent> publishedEvents = new CopyOnWriteArrayList<>();
+    InMemoryEvents eventBroker = new InMemoryEvents();
+    eventBroker.register(eventType, ce -> publishedEvents.add(ce));
+
+    try (WorkflowApplication app =
+        WorkflowApplication.builder()
+            .withEventConsumer(eventBroker)
+            .withEventPublisher(eventBroker)
+            .build()) {
+      app.workflowDefinition(workflow)
+          .instance(
+              List.of(
+                  new OrderName("item-1", "first"),
+                  new OrderName("item-2", "second"),
+                  new OrderName("item-3", "third")))
+          .start()
+          .join();
+      await()
+          .atMost(Duration.ofSeconds(2))
+          .pollInterval(Duration.ofMillis(10))
+          .until(() -> publishedEvents.size() == 3);
+
+      assertThat(
+              publishedEvents.stream()
+                  .map(CloudEvent::getData)
+                  .map(PojoCloudEventData.class::cast)
+                  .map(p -> p.getValue())
+                  .toList())
+          .isEqualTo(
+              List.of(
+                  Map.of("id", "item-1", "name", "first"),
+                  Map.of("id", "item-2", "name", "second"),
+                  Map.of("id", "item-3", "name", "third")));
+    }
+  }
+
   private static EnhancedOrder enhace(Order order) {
     return new EnhancedOrder(order.id(), 1000);
   }