AWS2 ddb-streams integration tests failures #2860

integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTest.java

@@ -16,9 +16,11 @@
  */
 package org.apache.camel.quarkus.component.aws2.ddb.it;
 
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import io.quarkus.test.common.QuarkusTestResource;
 import io.quarkus.test.junit.QuarkusTest;
@@ -40,9 +42,44 @@ class Aws2DdbTest {
 
     @Test
     public void crud() {
-
         final String key = "key" + UUID.randomUUID().toString().replace("-", "");
         final String msg = "val" + UUID.randomUUID().toString().replace("-", "");
+        final String initKeyPrefix = "initKey";
+        final String initMsg = "val";
+
+        /* Wait for the consumer to start. Test event has to be created periodically to ensure consumer reception */
+        RestAssured.given()
+                .contentType(ContentType.TEXT)
+                .body("")
+                .post("/aws2-ddb/item/" + initKeyPrefix + UUID.randomUUID().toString().replace("-", ""))
+                .then()
+                .statusCode(201);
+
+        /* Periodically check that init event is received and if nothing is received, invoke another one  */
+        Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+                () -> {
+                    ExtractableResponse<Response> result = RestAssured.get("/aws2-ddbstream/change")
+                            .then()
+                            .statusCode(200)
+                            .extract();
+
+                    LOG.info("Expecting at least 1 init event, got " + result.statusCode() + ": " + result.body().asString());
+
+                    List<Map> res = result.jsonPath().getList("$", Map.class);
+
+                    /* If there is no received event, consumer is still not running. Repeat insert. */
+                    if (res.isEmpty()) {
+                        RestAssured.given()
+                                .contentType(ContentType.TEXT)
+                                .body(initMsg)
+                                .post("/aws2-ddb/item/initKey" + UUID.randomUUID().toString().replace("-", ""))
+                                .then()
+                                .statusCode(201);
+                    }
+                    return res;
+                },
+                /* If at least one of the init events is recceived, consumer is working */
+                list -> !list.isEmpty());
 
         /* Ensure initially empty */
         RestAssured.get("/aws2-ddb/item/" + key)
@@ -110,9 +147,12 @@ public void crud() {
                             .extract();
 
                     LOG.info("Expecting 3 events got " + result.statusCode() + ": " + result.body().asString());
-                    return result.jsonPath().getList("$", Map.class);
+                    List<Map> retVal = result.jsonPath().getList("$", Map.class);
+                    //remove init events
+                    return retVal.stream().filter(m -> !String.valueOf(m.get("key")).startsWith("initKey"))
+                            .collect(Collectors.toList());
                 },
-                /* The above actions should trigger the following three change events */
+                /* The above actions should trigger the following three change events (initEvent is also present) */
                 list -> list.size() == 3
 
                         && key.equals(list.get(0).get("key"))
@@ -124,7 +164,6 @@ public void crud() {
 
                         && key.equals(list.get(2).get("key"))
                         && newMsg.equals(list.get(2).get("old")));
-
     }
 
 }