STX-221 Fix flaky parallel tests

.github/scripts/with-capture-docker-logs.sh

@@ -18,7 +18,8 @@ log_event() {
 # Named pipes so we can capture PIDs of both sides of each pipeline
 IMAGE_FIFO=$(mktemp -u)
 CONTAINER_FIFO=$(mktemp -u)
-mkfifo "$IMAGE_FIFO" "$CONTAINER_FIFO"
+NETWORK_FIFO=$(mktemp -u)
+mkfifo "$IMAGE_FIFO" "$CONTAINER_FIFO" "$NETWORK_FIFO"
 
 # --- Image pull logging ---
 docker events \
@@ -32,6 +33,18 @@ while read -r line; do
 done < "$IMAGE_FIFO" &
 IMAGE_READER_PID=$!
 
+# --- Network lifecycle logging ---
+docker events \
+  --filter 'type=network' \
+  --format '{{.Time}} NETWORK {{.Action}} name={{.Actor.Attributes.name}} id={{.Actor.ID}} container={{.Actor.Attributes.container}}' \
+  > "$NETWORK_FIFO" &
+NETWORK_EVENTS_PID=$!
+
+while read -r line; do
+  log_event "$line"
+done < "$NETWORK_FIFO" &
+NETWORK_READER_PID=$!
+
 # --- Container lifecycle logging ---
 docker events \
   --filter 'type=container' \
@@ -82,10 +95,12 @@ CONTAINER_READER_PID=$!
 cleanup() {
   log_event "Shutting down docker log collector"
   kill "$IMAGE_EVENTS_PID" "$IMAGE_READER_PID" \
+       "$NETWORK_EVENTS_PID" "$NETWORK_READER_PID" \
        "$CONTAINER_EVENTS_PID" "$CONTAINER_READER_PID" 2>/dev/null || true
   pkill -f "docker logs -f" 2>/dev/null || true
-  rm -f "$IMAGE_FIFO" "$CONTAINER_FIFO"
+  rm -f "$IMAGE_FIFO" "$CONTAINER_FIFO" "$NETWORK_FIFO"
   wait "$IMAGE_EVENTS_PID" "$IMAGE_READER_PID" \
+       "$NETWORK_EVENTS_PID" "$NETWORK_READER_PID" \
        "$CONTAINER_EVENTS_PID" "$CONTAINER_READER_PID" 2>/dev/null || true
 }
 trap cleanup EXIT

pom.xml

@@ -22,7 +22,7 @@
     <compiler-plugin.version>3.13.0</compiler-plugin.version>
     <checkstyle-plugin.version>3.6.0</checkstyle-plugin.version>
     <graalvm.version>25.0.2</graalvm.version>
-    <streamx.version>2.0.37</streamx.version>
+    <streamx.version>2.1.0-dev.STX-221.4</streamx.version>
     <cloudevents.version>4.0.1</cloudevents.version>
 
     <test.output.toFile>false</test.output.toFile>
@@ -167,6 +167,11 @@
       <version>${cloudevents.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>jakarta.ws.rs</groupId>
+      <artifactId>jakarta.ws.rs-api</artifactId>
+    </dependency>
+
     <!-- Native image dependencies -->
     <dependency>
       <groupId>org.graalvm.sdk</groupId>
@@ -586,4 +591,4 @@
       </build>
     </profile>
   </profiles>
-</project>
+</project>

src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java

@@ -16,6 +16,7 @@
 import io.quarkus.test.junit.QuarkusTest;
 import java.nio.file.Paths;
 import java.time.Duration;
+import java.util.List;
 import java.util.UUID;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.AfterEach;
@@ -26,17 +27,21 @@
 @DisabledIfDockerUnavailable
 public class RunCommandIT extends CliBaseIT {
 
-  private static final String PREFIX =
-      "sx-run-" + UUID.randomUUID().toString().substring(0, 4) + "-";
+  private static String freshPrefix() {
+    return "sx-run-" + UUID.randomUUID().toString().substring(0, 4) + "-";
+  }
 
   private static final String WEB_SERVER_SINK_IMAGE =
       "ghcr.io/streamx-com/streamx-blueprints/web-server-sink:3.0.7-jvm";
 
   private static final String BLOCKER_IMAGE = "alpine:3.20";
 
+  private static final int SIGTERM_EXIT_CODE = 128 + 15;
+
   @BeforeEach
   void isolateRunFromConcurrentInstances() {
-    System.setProperty("streamx.container.startup-timeout-seconds", "180");
+    System.setProperty("streamx.container.startup-timeout-seconds",
+        MeshTestSupport.CONTAINER_STARTUP_TIMEOUT_SECONDS);
     System.setProperty("streamx.runner.pulsar.broker-port",
         String.valueOf(MeshTestSupport.freePort()));
     System.setProperty("streamx.runner.pulsar.http-port",
@@ -51,6 +56,7 @@ void stopMeshAndResetRunnerState() {
     } catch (Exception ignored) {
       // best-effort cleanup
     }
+    awaitAsyncCommands();
     System.clearProperty("streamx.runner.mesh-name-prefix");
     System.clearProperty("streamx.container.startup-timeout-seconds");
     System.clearProperty("streamx.runner.pulsar.broker-port");
@@ -60,7 +66,7 @@ void stopMeshAndResetRunnerState() {
 
   @Test
   void shouldWarnWhenEnvVariableIsUndefined() throws Exception {
-    System.setProperty("streamx.runner.mesh-name-prefix", PREFIX);
+    System.setProperty("streamx.runner.mesh-name-prefix", freshPrefix());
     exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE);
     exec("settings", "unset", "STREAMX_OWNER_SERVICE_NAME");
     clearEnv("STREAMX_OWNER_SERVICE_NAME");
@@ -73,18 +79,14 @@ void shouldWarnWhenEnvVariableIsUndefined() throws Exception {
     AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath);
 
     try {
-      Awaitility.await()
-          .atMost(Duration.ofMinutes(3))
-          .pollInterval(Duration.ofSeconds(1))
-          .until(() -> handle.getStderr()
-              .contains("Environment variable 'STREAMX_OWNER_SERVICE_NAME'"));
+      awaitStderrContains(handle, "Environment variable 'STREAMX_OWNER_SERVICE_NAME'");
 
       assertThat(handle.getStderr())
           .contains("WARNING:")
           .contains("STREAMX_OWNER_SERVICE_NAME");
     } finally {
-      if (handle.thread().isAlive()) {
-        handle.interruptAndJoin(Duration.ofSeconds(30).toMillis());
+      if (handle.isAlive()) {
+        handle.stopAndJoin(Duration.ofSeconds(30).toMillis());
       }
     }
   }
@@ -101,9 +103,10 @@ void shouldFailWhenMeshFileDoesNotExist() throws Exception {
 
   @Test
   void shouldReportContainerFailureWhenItsHostPortIsAlreadyTaken() throws Exception {
-    System.setProperty("streamx.runner.mesh-name-prefix", PREFIX);
+    String prefix = freshPrefix();
+    System.setProperty("streamx.runner.mesh-name-prefix", prefix);
     exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE);
-    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner");
+    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", prefix + "test-owner");
 
     String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml")
         .toAbsolutePath()
@@ -114,27 +117,32 @@ void shouldReportContainerFailureWhenItsHostPortIsAlreadyTaken() throws Exceptio
     String blockerId = startPortBlocker(blockedPort);
     System.setProperty("test.proxy.host-port", String.valueOf(blockedPort));
 
-    AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath);
+    AsyncProcessHandle handle = execAsync("local", "run", "--verbose", "-f=" + meshPath);
     try {
-      Awaitility.await()
-          .atMost(Duration.ofMinutes(3))
-          .pollInterval(Duration.ofSeconds(1))
-          .untilAsserted(() -> {
-            assertThat(handle.getStdout())
-                .as("the user must be told which container failed")
-                .contains("rest-ingestion.proxy failed");
-            assertThat(handle.getStderr())
-                .as("the run must be reported as failed")
-                .contains(msg.somethingWentWrong().strip());
-          });
+      awaitStderrContains(handle, msg.somethingWentWrong().strip(),
+          String.valueOf(blockedPort));
     } finally {
-      if (handle.thread().isAlive()) {
-        handle.interruptAndJoin(Duration.ofSeconds(30).toMillis());
+      if (handle.isAlive()) {
+        handle.stopAndJoin(Duration.ofSeconds(30).toMillis());
       }
       removePortBlocker(blockerId);
     }
   }
 
+  private static void awaitNoContainersWithPrefix(String prefix) throws Exception {
+    try (DockerClient docker = DockerClientFactory.create()) {
+      Awaitility.await()
+          .atMost(Duration.ofSeconds(60))
+          .pollInterval(Duration.ofSeconds(1))
+          .untilAsserted(() -> assertThat(docker.listContainersCmd()
+              .withShowAll(true)
+              .withNameFilter(List.of(prefix))
+              .exec())
+              .as("mesh containers with prefix %s must be removed after the run stops", prefix)
+              .isEmpty());
+    }
+  }
+
   private static String startPortBlocker(int hostPort) throws Exception {
     try (DockerClient docker = DockerClientFactory.create()) {
       docker.pullImageCmd(BLOCKER_IMAGE).start().awaitCompletion();
@@ -162,9 +170,10 @@ private static void removePortBlocker(String containerId) {
 
   @Test
   void shouldFailWhenSystemPropertyIsUndefined() throws Exception {
-    System.setProperty("streamx.runner.mesh-name-prefix", PREFIX);
+    String prefix = freshPrefix();
+    System.setProperty("streamx.runner.mesh-name-prefix", prefix);
     exec("settings", "unset", "config.image.interpolated");
-    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner");
+    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", prefix + "test-owner");
 
     String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml")
         .toAbsolutePath()
@@ -177,16 +186,16 @@ void shouldFailWhenSystemPropertyIsUndefined() throws Exception {
       Awaitility.await()
           .atMost(Duration.ofMinutes(3))
           .pollInterval(Duration.ofSeconds(1))
-          .until(() -> !handle.thread().isAlive());
+          .until(() -> !handle.isAlive());
 
       ProcessResult result = handle.toResult();
       assertThat(result.exitCode()).isNotEqualTo(0);
       assertThat(result.stderr())
           .contains("Property 'config.image.interpolated'")
           .contains("is not set");
     } finally {
-      if (handle.thread().isAlive()) {
-        handle.interruptAndJoin(Duration.ofSeconds(30).toMillis());
+      if (handle.isAlive()) {
+        handle.stopAndJoin(Duration.ofSeconds(30).toMillis());
       }
     }
   }
@@ -219,27 +228,26 @@ void shouldBridgeRunnerSettingToSystemPropertyForLocalRun() throws Exception {
     AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath);
 
     try {
-      Awaitility.await()
-          .atMost(Duration.ofMinutes(3))
-          .pollInterval(Duration.ofSeconds(1))
-          .until(() -> handle.getStdout().contains("STREAMX IS READY!"));
+      awaitStdoutContains(handle, "STREAMX IS READY!");
 
       assertThat(handle.getStdout())
           .as("runner should use the prefix from streamxHome settings via the bridge")
           .contains(bridgedPrefix);
     } finally {
-      if (handle.thread().isAlive()) {
-        handle.interruptAndJoin(Duration.ofSeconds(30).toMillis());
+      if (handle.isAlive()) {
+        handle.stopAndJoin(Duration.ofSeconds(30).toMillis());
       }
+      exec("settings", "unset", "streamx.runner.mesh-name-prefix");
       System.clearProperty("streamx.runner.mesh-name-prefix");
     }
   }
 
   @Test
   void shouldSucceedWhenInterpolationValuesAreDefined() throws Exception {
-    System.setProperty("streamx.runner.mesh-name-prefix", PREFIX);
+    String prefix = freshPrefix();
+    System.setProperty("streamx.runner.mesh-name-prefix", prefix);
     exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE);
-    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner");
+    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", prefix + "test-owner");
 
     String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml")
         .toAbsolutePath()
@@ -249,33 +257,28 @@ void shouldSucceedWhenInterpolationValuesAreDefined() throws Exception {
     AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath);
 
     try {
-      Awaitility.await()
-          .atMost(Duration.ofMinutes(3))
-          .pollInterval(Duration.ofSeconds(1))
-          .until(() -> handle.getStdout().contains("STREAMX IS READY!"));
+      awaitStdoutContains(handle, "STREAMX IS READY!");
 
       Thread.sleep(Duration.ofSeconds(5));
-      assertThat(handle.thread().isAlive()).isTrue();
+      assertThat(handle.isAlive()).isTrue();
 
-      handle.interruptAndJoin(Duration.ofSeconds(30).toMillis());
-      assertThat(handle.thread().isAlive()).isFalse();
+      handle.stopAndJoin(Duration.ofSeconds(30).toMillis());
+      assertThat(handle.isAlive()).isFalse();
 
       ProcessResult result = handle.toResult();
-      result.assertSuccess();
-      assertThat(result.stdout()).contains("Stopping mesh...");
-      assertThat(result.stderr()).doesNotContain("Exception");
+      assertThat(result.exitCode()).isIn(0, SIGTERM_EXIT_CODE);
+      awaitNoContainersWithPrefix(prefix);
     } finally {
-      if (handle.thread().isAlive()) {
-        handle.interruptAndJoin(Duration.ofSeconds(30).toMillis());
+      if (handle.isAlive()) {
+        handle.stopAndJoin(Duration.ofSeconds(30).toMillis());
       }
     }
   }
 
   @Test
   void shouldStartMeshSecondTimeAfterPreviousStopped() throws Exception {
-    System.setProperty("streamx.runner.mesh-name-prefix", PREFIX);
     exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE);
-    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner");
+    exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", freshPrefix() + "test-owner");
 
     String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml")
         .toAbsolutePath()
@@ -286,19 +289,22 @@ void shouldStartMeshSecondTimeAfterPreviousStopped() throws Exception {
     runUntilReadyThenStop(meshPath);
   }
 
-  private void runUntilReadyThenStop(String meshPath) throws InterruptedException {
+  private void runUntilReadyThenStop(String meshPath) throws Exception {
+    System.setProperty("streamx.runner.mesh-name-prefix", freshPrefix());
+    System.setProperty("streamx.runner.pulsar.broker-port",
+        String.valueOf(MeshTestSupport.freePort()));
+    System.setProperty("streamx.runner.pulsar.http-port",
+        String.valueOf(MeshTestSupport.freePort()));
+    System.setProperty("test.proxy.host-port", String.valueOf(MeshTestSupport.freePort()));
     AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath);
     try {
-      Awaitility.await()
-          .atMost(Duration.ofMinutes(3))
-          .pollInterval(Duration.ofSeconds(1))
-          .until(() -> handle.getStdout().contains("STREAMX IS READY!"));
+      awaitStdoutContains(handle, "STREAMX IS READY!");
 
       assertThat(handle.getStderr())
           .doesNotContain("MissingReflectionRegistrationError");
     } finally {
-      if (handle.thread().isAlive()) {
-        handle.interruptAndJoin(Duration.ofSeconds(30).toMillis());
+      if (handle.isAlive()) {
+        handle.stopAndJoin(Duration.ofSeconds(30).toMillis());
       }
     }
   }