From d5124168a98b1ffed9df6514030dbe20af254e5b Mon Sep 17 00:00:00 2001
From: Timon Borter <timon.borter@gmx.ch>
Date: Sun, 20 Oct 2024 18:11:44 +0200
Subject: [PATCH] ci(#1234): fix flaky kafka selective xml test

---
 .../integration/KafkaEndpointJavaIT.java      | 14 +----
 .../kafka/integration/KafkaEndpointXmlIT.java |  2 +-
 .../KafkaEndpointIT_selectiveMessage.xml      | 63 +++++++++++--------
 3 files changed, 40 insertions(+), 39 deletions(-)

diff --git a/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointJavaIT.java b/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointJavaIT.java
index 2b5abb4b42..83e448ef61 100644
--- a/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointJavaIT.java
+++ b/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointJavaIT.java
@@ -16,8 +16,6 @@
 
 package org.citrusframework.kafka.integration;
 
-import java.time.Duration;
-
 import org.assertj.core.api.ThrowableAssert;
 import org.citrusframework.annotations.CitrusTest;
 import org.citrusframework.exceptions.CitrusRuntimeException;
@@ -29,6 +27,8 @@
 import org.citrusframework.testng.spring.TestNGCitrusSpringSupport;
 import org.testng.annotations.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.citrusframework.actions.ReceiveMessageAction.Builder.receive;
 import static org.citrusframework.actions.SendMessageAction.Builder.send;
@@ -39,7 +39,7 @@
 import static org.citrusframework.kafka.endpoint.selector.KafkaMessageByHeaderSelector.kafkaHeaderContains;
 import static org.citrusframework.kafka.endpoint.selector.KafkaMessageByHeaderSelector.kafkaHeaderEquals;
 
-@Test
+@Test(singleThreaded = true)
 public class KafkaEndpointJavaIT extends TestNGCitrusSpringSupport {
 
     @BindToRegistry
@@ -48,7 +48,6 @@ public class KafkaEndpointJavaIT extends TestNGCitrusSpringSupport {
             .topic("hello")
             .build();
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_headerEquals_citrus_DSL() {
         var body = "findKafkaEvent_headerEquals_citrus_DSL";
@@ -74,7 +73,6 @@ public void findKafkaEvent_headerEquals_citrus_DSL() {
         );
     }
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_headerContains_citrus_DSL() {
         var body = "findKafkaEvent_headerContains_citrus_DSL";
@@ -100,7 +98,6 @@ public void findKafkaEvent_headerContains_citrus_DSL() {
         );
     }
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_headerStartsWith_citrus_DSL() {
         var body = "findKafkaEvent_headerStartsWith_citrus_DSL";
@@ -132,7 +129,6 @@ public void findKafkaEvent_headerStartsWith_citrus_DSL() {
         );
     }
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_headerEndsWith_citrus_DSL() {
         var body = "findKafkaEvent_headerEndsWith_citrus_DSL";
@@ -164,7 +160,6 @@ public void findKafkaEvent_headerEndsWith_citrus_DSL() {
         );
     }
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_nothingFound_noMatch_citrus_DSL() {
         var body = "findKafkaEvent_nothingFound_noMatch_citrus_DSL";
@@ -195,7 +190,6 @@ public void findKafkaEvent_nothingFound_noMatch_citrus_DSL() {
                 .hasMessageContaining("Failed to resolve Kafka message using selector");
     }
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_nothingFound_outsideLookbackWindow_citrus_DSL() {
         var body = "findKafkaEvent_nothingFound_outsideLookbackWindow_citrus_DSL";
@@ -228,7 +222,6 @@ public void findKafkaEvent_nothingFound_outsideLookbackWindow_citrus_DSL() {
                 .hasMessageContaining("Failed to resolve Kafka message using selector");
     }
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_duplicateEntriesFound_citrus_DSL() {
         var body = "findKafkaEvent_duplicateEntriesFound_citrus_DSL";
@@ -263,7 +256,6 @@ public void findKafkaEvent_duplicateEntriesFound_citrus_DSL() {
                 .hasMessageContaining("More than one matching record found in topic");
     }
 
-    @Test
     @CitrusTest
     public void findKafkaEvent_headerEquals_java_DSL() {
         var body = "findKafkaEvent_headerEquals_java_DSL";
diff --git a/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointXmlIT.java b/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointXmlIT.java
index 8099e6a3ad..17661b07ce 100644
--- a/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointXmlIT.java
+++ b/endpoints/citrus-kafka/src/test/java/org/citrusframework/kafka/integration/KafkaEndpointXmlIT.java
@@ -21,7 +21,7 @@
 import org.citrusframework.testng.spring.TestNGCitrusSpringSupport;
 import org.testng.annotations.Test;
 
-@Test
+@Test(singleThreaded = true)
 public class KafkaEndpointXmlIT extends TestNGCitrusSpringSupport {
 
     @CitrusTestSource(type = TestLoader.SPRING, name = "KafkaEndpointIT_singleMessage")
diff --git a/endpoints/citrus-kafka/src/test/resources/org/citrusframework/kafka/integration/KafkaEndpointIT_selectiveMessage.xml b/endpoints/citrus-kafka/src/test/resources/org/citrusframework/kafka/integration/KafkaEndpointIT_selectiveMessage.xml
index 3e97cd2992..2b9c16a424 100644
--- a/endpoints/citrus-kafka/src/test/resources/org/citrusframework/kafka/integration/KafkaEndpointIT_selectiveMessage.xml
+++ b/endpoints/citrus-kafka/src/test/resources/org/citrusframework/kafka/integration/KafkaEndpointIT_selectiveMessage.xml
@@ -55,33 +55,42 @@
         </header>
       </send>
 
-      <receive endpoint="helloKafkaEndpoint">
-        <description>Receive Kafka request: Kafka broker -> Citrus</description>
-        <selector>
-          <element name="header-filter-key" value="traceparent"/>
-          <element name="header-filter-value" value="${traceId}"/>
-          <element name="header-filter-comparator" value="CONTAINS"/>
-          <element name="event-lookback-window" value="PT1S"/>
-        </selector>
-        <message>
-          <data>
-            <![CDATA[
-              <HelloRequest xmlns="http://citrusframework.org/schemas/samples/HelloService.xsd">
-                 <MessageId>${messageId}</MessageId>
-                 <CorrelationId>${correlationId}</CorrelationId>
-                 <User>${user}</User>
-                 <Text>Hello Citrus</Text>
-               </HelloRequest>
-            ]]>
-          </data>
-        </message>
-        <header>
-          <element name="Operation" value="sayHello"/>
-          <element name="citrus_kafka_topic" value="hello"/>
-          <element name="citrus_kafka_partition" value="0"/>
-          <element name="citrus_kafka_offset" value="@assertThat(greaterThanOrEqualTo(0))@"/>
-        </header>
-      </receive>
+      <create-variables>
+        <variable name="extractedTraceparent"></variable>
+      </create-variables>
+
+      <repeat-onerror-until-true condition="@assertThat('${extractedTraceparent}', 'equalTo(01-${traceId}-${spanId}-00)')@">
+        <receive endpoint="helloKafkaEndpoint">
+          <description>Receive Kafka request: Kafka broker -> Citrus</description>
+          <selector>
+            <element name="header-filter-key" value="traceparent"/>
+            <element name="header-filter-value" value="${traceId}"/>
+            <element name="header-filter-comparator" value="CONTAINS"/>
+            <element name="event-lookback-window" value="PT1S"/>
+          </selector>
+          <message>
+            <data>
+              <![CDATA[
+                <HelloRequest xmlns="http://citrusframework.org/schemas/samples/HelloService.xsd">
+                   <MessageId>${messageId}</MessageId>
+                   <CorrelationId>${correlationId}</CorrelationId>
+                   <User>${user}</User>
+                   <Text>Hello Citrus</Text>
+                 </HelloRequest>
+              ]]>
+            </data>
+          </message>
+          <header>
+            <element name="Operation" value="sayHello"/>
+            <element name="citrus_kafka_topic" value="hello"/>
+            <element name="citrus_kafka_partition" value="0"/>
+            <element name="citrus_kafka_offset" value="@assertThat(greaterThanOrEqualTo(0))@"/>
+          </header>
+          <extract>
+            <header name="traceparent" variable="${extractedTraceparent}"/>
+          </extract>
+        </receive>
+      </repeat-onerror-until-true>
     </actions>
   </testcase>
 </spring:beans>