From bd5c66b24486b762b12df79bcea40827ddc7aa81 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 11:11:56 +0300 Subject: [PATCH 01/29] STX-221 Fix flaky parallel tests --- .../com/streamx/cli/test/MeshTestSupport.java | 37 +++++++++++++++++-- 1 file changed, 34 insertions(+), 3 deletions(-) diff --git a/src/test/java/com/streamx/cli/test/MeshTestSupport.java b/src/test/java/com/streamx/cli/test/MeshTestSupport.java index aa2b646f..31ff897f 100644 --- a/src/test/java/com/streamx/cli/test/MeshTestSupport.java +++ b/src/test/java/com/streamx/cli/test/MeshTestSupport.java @@ -5,6 +5,8 @@ import java.io.IOException; import java.io.PrintStream; import java.net.ServerSocket; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.util.UUID; @@ -22,6 +24,9 @@ */ public final class MeshTestSupport { + private static final Path PORT_RESERVATIONS = + Paths.get(System.getProperty("java.io.tmpdir"), "streamx-cli-test-ports"); + private static volatile MeshManager activeMeshManager; private static volatile int activeProxyPort; private static volatile int activePulsarHttpPort; @@ -31,11 +36,37 @@ public final class MeshTestSupport { private MeshTestSupport() { } + /** + * The port reservation machinery is needed to fix parallel tests flakiness for commands which + * run docker containers like `streamx local run`. + * The naive new ServerSocket(0) doesn't work reliably in this case. + */ public static int freePort() { - try (ServerSocket s = new ServerSocket(0)) { - return s.getLocalPort(); + for (int attempt = 0; attempt < 100; attempt++) { + int candidate; + try (ServerSocket s = new ServerSocket(0)) { + candidate = s.getLocalPort(); + } catch (IOException e) { + throw new RuntimeException("Failed to find a free port", e); + } + if (reserve(candidate)) { + return candidate; + } + } + throw new IllegalStateException("Could not reserve a free port after 100 attempts"); + } + + private static boolean reserve(int port) { + try { + Path lock = PORT_RESERVATIONS.resolve(port + ".lock"); + Files.createDirectories(PORT_RESERVATIONS); + Files.createFile(lock); + lock.toFile().deleteOnExit(); + return true; + } catch (FileAlreadyExistsException takenByAnotherJvm) { + return false; } catch (IOException e) { - throw new RuntimeException("Failed to find a free port", e); + throw new RuntimeException("Failed to reserve port " + port, e); } } From 17d771bdac8648bb9643ff72e1dcbadfcb274723 Mon Sep 17 00:00:00 2001 From: "streamx-cli-release-bot[bot]" <264353404+streamx-cli-release-bot[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:16:39 +0000 Subject: [PATCH 02/29] [no ci] [maven-release-plugin] prepare release 2.0.4-rc.bd5c66b --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 711a143a..0cbf847f 100644 --- a/pom.xml +++ b/pom.xml @@ -4,13 +4,13 @@ com.streamx.cli streamx-cli - 2.0.4-SNAPSHOT + 2.0.4-rc.bd5c66b scm:git:https://github.com/streamx-com/streamx-cli.git scm:git:https://github.com/streamx-com/streamx-cli.git https://github.com/streamx-com/streamx-cli - HEAD + 2.0.4-rc.bd5c66b From 15abeff4172b483e23615f9fdc36795d4e0fd97d Mon Sep 17 00:00:00 2001 From: "streamx-cli-release-bot[bot]" <264353404+streamx-cli-release-bot[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:16:39 +0000 Subject: [PATCH 03/29] [no ci] [maven-release-plugin] prepare for next development iteration --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 0cbf847f..711a143a 100644 --- a/pom.xml +++ b/pom.xml @@ -4,13 +4,13 @@ com.streamx.cli streamx-cli - 2.0.4-rc.bd5c66b + 2.0.4-SNAPSHOT scm:git:https://github.com/streamx-com/streamx-cli.git scm:git:https://github.com/streamx-com/streamx-cli.git https://github.com/streamx-com/streamx-cli - 2.0.4-rc.bd5c66b + HEAD From 0d681b8675e20a3ebe1fd0ca61dce9b8a2f1b2a7 Mon Sep 17 00:00:00 2001 From: "streamx-cli-release-bot[bot]" <264353404+streamx-cli-release-bot[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:48:59 +0000 Subject: [PATCH 04/29] [no ci] [maven-release-plugin] prepare release 2.0.4-rc.15abeff --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 711a143a..58b697a6 100644 --- a/pom.xml +++ b/pom.xml @@ -4,13 +4,13 @@ com.streamx.cli streamx-cli - 2.0.4-SNAPSHOT + 2.0.4-rc.15abeff scm:git:https://github.com/streamx-com/streamx-cli.git scm:git:https://github.com/streamx-com/streamx-cli.git https://github.com/streamx-com/streamx-cli - HEAD + 2.0.4-rc.15abeff From f9e8ec2ec0e2e6ee70faa30d31ee8e446669c55e Mon Sep 17 00:00:00 2001 From: "streamx-cli-release-bot[bot]" <264353404+streamx-cli-release-bot[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:48:59 +0000 Subject: [PATCH 05/29] [no ci] [maven-release-plugin] prepare for next development iteration --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 58b697a6..711a143a 100644 --- a/pom.xml +++ b/pom.xml @@ -4,13 +4,13 @@ com.streamx.cli streamx-cli - 2.0.4-rc.15abeff + 2.0.4-SNAPSHOT scm:git:https://github.com/streamx-com/streamx-cli.git scm:git:https://github.com/streamx-com/streamx-cli.git https://github.com/streamx-com/streamx-cli - 2.0.4-rc.15abeff + HEAD From 1470b912a51e685a1cf3122f14b5d0d4fa0b59a7 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 11:58:53 +0300 Subject: [PATCH 06/29] Force CI build From 078c6840e55b1cf15d90e86de5211a1511cd343b Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 11:59:52 +0300 Subject: [PATCH 07/29] Force CI build From 5be869983442ae32d608c91bfbd3f503e47f8910 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 11:59:58 +0300 Subject: [PATCH 08/29] Force CI build From 482a3065db814a04f07050dbbeb7458aae1242ed Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 12:00:07 +0300 Subject: [PATCH 09/29] Force CI build From 71e40aee32acb85bd0af215136eb98426a5083ef Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 12:00:13 +0300 Subject: [PATCH 10/29] Force CI build From d57b09f7c125931c153e2252400616b9871f0420 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 12:00:23 +0300 Subject: [PATCH 11/29] Force CI build From 937927c73fc94d9eae638a0f73d2c5d387fc714f Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 12:00:28 +0300 Subject: [PATCH 12/29] Force CI build From b8ec72f2ff43bce5e6f029b373b03a77533667e3 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 12:39:39 +0300 Subject: [PATCH 13/29] STX-221 Fix flaky parallel tests --- .../com/streamx/cli/commands/local/run/RunCommandIT.java | 3 ++- src/test/java/com/streamx/cli/test/MeshTestSupport.java | 5 ++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java b/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java index 23c3b653..5c4b0e00 100644 --- a/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java +++ b/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java @@ -36,7 +36,8 @@ public class RunCommandIT extends CliBaseIT { @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", diff --git a/src/test/java/com/streamx/cli/test/MeshTestSupport.java b/src/test/java/com/streamx/cli/test/MeshTestSupport.java index 31ff897f..2bd9816b 100644 --- a/src/test/java/com/streamx/cli/test/MeshTestSupport.java +++ b/src/test/java/com/streamx/cli/test/MeshTestSupport.java @@ -27,6 +27,8 @@ public final class MeshTestSupport { private static final Path PORT_RESERVATIONS = Paths.get(System.getProperty("java.io.tmpdir"), "streamx-cli-test-ports"); + public static final String CONTAINER_STARTUP_TIMEOUT_SECONDS = "300"; + private static volatile MeshManager activeMeshManager; private static volatile int activeProxyPort; private static volatile int activePulsarHttpPort; @@ -93,7 +95,8 @@ public static void startMesh(String meshYamlPath) { "http://localhost:" + activeProxyPort); System.setProperty("test.proxy.host-port", String.valueOf(activeProxyPort)); - System.setProperty("streamx.container.startup-timeout-seconds", "180"); + System.setProperty("streamx.container.startup-timeout-seconds", + CONTAINER_STARTUP_TIMEOUT_SECONDS); capturedToken = null; tokenLatch = new CountDownLatch(1); From 4cb765177d043ef4e7e40848f7833ef794fc28d4 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 12:39:46 +0300 Subject: [PATCH 14/29] Force CI build From 7c061c2dcf39c717c416d9c80e100554dbd7c5ef Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Wed, 29 Jul 2026 12:39:55 +0300 Subject: [PATCH 15/29] Force CI build From 1af751bf5ee392934508455b5b1d900b283364d7 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:07:00 +0300 Subject: [PATCH 16/29] STX-221 Fix flaky parallel tests and harden the local-run test setup --- .github/scripts/with-capture-docker-logs.sh | 19 +- pom.xml | 9 +- .../cli/commands/local/run/RunCommandIT.java | 125 ++++----- .../java/com/streamx/cli/test/CliBaseIT.java | 240 ++++++++++++++---- .../com/streamx/cli/test/MeshTestSupport.java | 34 ++- 5 files changed, 298 insertions(+), 129 deletions(-) diff --git a/.github/scripts/with-capture-docker-logs.sh b/.github/scripts/with-capture-docker-logs.sh index 7482bf71..4e75dd5f 100755 --- a/.github/scripts/with-capture-docker-logs.sh +++ b/.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 diff --git a/pom.xml b/pom.xml index 711a143a..97cf9aec 100644 --- a/pom.xml +++ b/pom.xml @@ -22,7 +22,7 @@ 3.13.0 3.6.0 25.0.2 - 2.0.37 + 2.1.0-dev.STX-221.4 4.0.1 false @@ -167,6 +167,11 @@ ${cloudevents.version} + + jakarta.ws.rs + jakarta.ws.rs-api + + org.graalvm.sdk @@ -586,4 +591,4 @@ - \ No newline at end of file + diff --git a/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java b/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java index 5c4b0e00..b65191a2 100644 --- a/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java +++ b/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,14 +27,17 @@ @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", @@ -52,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"); @@ -61,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"); @@ -74,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()); } } } @@ -102,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() @@ -115,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(); @@ -163,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() @@ -178,7 +186,7 @@ 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); @@ -186,8 +194,8 @@ void shouldFailWhenSystemPropertyIsUndefined() throws Exception { .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()); } } } @@ -220,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() @@ -250,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() @@ -287,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()); } } } diff --git a/src/test/java/com/streamx/cli/test/CliBaseIT.java b/src/test/java/com/streamx/cli/test/CliBaseIT.java index ba49fc2f..ae69b2a0 100644 --- a/src/test/java/com/streamx/cli/test/CliBaseIT.java +++ b/src/test/java/com/streamx/cli/test/CliBaseIT.java @@ -1,5 +1,7 @@ package com.streamx.cli.test; +import static org.assertj.core.api.Assertions.assertThat; + import com.streamx.cli.commands.StreamxCommand; import com.streamx.cli.framework.AbstractCommand; import io.quarkus.arc.Arc; @@ -11,14 +13,25 @@ import java.io.InputStream; import java.io.OutputStream; import java.io.PrintStream; +import java.io.UncheckedIOException; +import java.lang.management.ManagementFactory; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.time.Duration; import java.util.ArrayList; +import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Stream; +import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; @@ -39,6 +52,14 @@ public abstract class CliBaseIT { private Process process; private final Map envVars = new HashMap<>(); + private static final long ASYNC_COMMAND_SHUTDOWN_TIMEOUT_MILLIS = 60_000; + + private static final Path BUILD_OUTPUT_DIR = Path.of("target"); + + private static final List ASYNC_COMMANDS = new ArrayList<>(); + + private static final AtomicInteger ASYNC_COMMAND_COUNTER = new AtomicInteger(); + private static boolean isNative() { return "true".equals(System.getProperty("native.image")); } @@ -56,17 +77,15 @@ static void ensureBuilt() { } protected void setEnv(String key, String value) { - if (isNative()) { - envVars.put(key, value); - } else { + envVars.put(key, value); + if (!isNative()) { System.setProperty(key, value); } } protected void clearEnv(String key) { - if (isNative()) { - envVars.remove(key); - } else { + envVars.remove(key); + if (!isNative()) { System.clearProperty(key); } } @@ -231,9 +250,13 @@ private ProcessResult execSubprocess( return new ProcessResult(process.exitValue(), stdout, stderr); } - private record StreamCapture(Thread thread, ByteArrayOutputStream buffer) { + record StreamCapture(Thread thread, ByteArrayOutputStream buffer) { String join() throws InterruptedException { thread.join(); + return content(); + } + + String content() { return buffer.toString(StandardCharsets.UTF_8); } } @@ -271,81 +294,188 @@ public void assertExitCode(int expected) { } public record AsyncProcessHandle( - Thread thread, - ByteArrayOutputStream stdout, - ByteArrayOutputStream stderr, - AtomicInteger exitCode + Process process, + StreamCapture stdoutCapture, + StreamCapture stderrCapture ) { public String getStdout() { - return stdout.toString(StandardCharsets.UTF_8); + return stdoutCapture.content(); } public String getStderr() { - return stderr.toString(StandardCharsets.UTF_8); + return stderrCapture.content(); + } + + public boolean isAlive() { + return process.isAlive(); + } + + public void stopAndJoin(long timeoutMillis) throws InterruptedException { + process.destroy(); + if (!process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS)) { + process.destroyForcibly(); + process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS); + } + joinCaptures(timeoutMillis); } - public void interruptAndJoin(long timeoutMillis) throws InterruptedException { - thread.interrupt(); - thread.join(timeoutMillis); + public ProcessResult toResult() throws InterruptedException { + if (!process.isAlive()) { + joinCaptures(ASYNC_COMMAND_SHUTDOWN_TIMEOUT_MILLIS); + } + return new ProcessResult( + process.isAlive() ? -1 : process.exitValue(), getStdout(), getStderr()); } - public ProcessResult toResult() { - return new ProcessResult(exitCode.get(), getStdout(), getStderr()); + private void joinCaptures(long timeoutMillis) throws InterruptedException { + stdoutCapture.thread().join(timeoutMillis); + stderrCapture.thread().join(timeoutMillis); } } - protected AsyncProcessHandle execAsync(String... args) { - ByteArrayOutputStream out = new ByteArrayOutputStream(); - ByteArrayOutputStream err = new ByteArrayOutputStream(); - AtomicInteger exitCode = new AtomicInteger(-1); + protected AsyncProcessHandle execAsync(String... args) throws IOException { + awaitAsyncCommands(); - PrintStream originalOut = System.out; - PrintStream originalErr = System.err; + List command = cliLaunchCommand(); + command.addAll(List.of(args)); - PrintStream teeOut = new PrintStream(new TeeOutputStream(out, originalOut), true); - PrintStream teeErr = new PrintStream(new TeeOutputStream(err, originalErr), true); + ProcessBuilder pb = new ProcessBuilder(command); + pb.environment().put("STREAMX_HOME", streamxHome.toAbsolutePath().toString()); + pb.environment().putAll(envVars); + Process process = pb.start(); + + AsyncProcessHandle handle = new AsyncProcessHandle( + process, + captureAndForward(process.getInputStream(), System.out), + captureAndForward(process.getErrorStream(), System.err)); + ASYNC_COMMANDS.add(handle); + return handle; + } - Thread thread = Thread.ofVirtual().start(() -> { - System.setOut(teeOut); - System.setErr(teeErr); - System.setProperty("STREAMX_HOME", streamxHome.toAbsolutePath().toString()); + protected void awaitAsyncCommands() { + for (AsyncProcessHandle handle : ASYNC_COMMANDS) { try { - exitCode.set(createCommandLine().execute(args)); - } finally { - System.clearProperty("STREAMX_HOME"); - System.setOut(originalOut); - System.setErr(originalErr); + handle.stopAndJoin(ASYNC_COMMAND_SHUTDOWN_TIMEOUT_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; } - }); + } + ASYNC_COMMANDS.clear(); + } - return new AsyncProcessHandle(thread, out, err, exitCode); + protected void awaitStdoutContains(AsyncProcessHandle handle, String... expected) { + awaitOutputContains(handle, AsyncProcessHandle::getStdout, expected); } - private static class TeeOutputStream extends OutputStream { - private final OutputStream buffer; - private final OutputStream console; + protected void awaitStderrContains(AsyncProcessHandle handle, String... expected) { + awaitOutputContains(handle, AsyncProcessHandle::getStderr, expected); + } - TeeOutputStream(OutputStream buffer, OutputStream console) { - this.buffer = buffer; - this.console = console; + private static void awaitOutputContains( + AsyncProcessHandle handle, + Function output, + String... expected + ) { + try { + Awaitility.await() + .atMost(Duration.ofMinutes(3)) + .pollInterval(Duration.ofSeconds(1)) + .failFast(() -> { + if (handle.isAlive()) { + return false; + } + try { + handle.joinCaptures(Duration.ofSeconds(5).toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return !containsAll(output.apply(handle), expected); + }) + .untilAsserted(() -> assertThat(output.apply(handle)).contains(expected)); + } catch (RuntimeException e) { + throw new AssertionError( + ("Expected the CLI output to contain %s. CLI process alive: %s" + + "%n--- captured stdout ---%n%s%n--- captured stderr ---%n%s%s") + .formatted(List.of(expected), handle.isAlive(), + handle.getStdout(), handle.getStderr(), errorLogContent(handle)), + e); } + } - @Override - public void write(int b) throws IOException { - buffer.write(b); - console.write(b); + private static String errorLogContent(AsyncProcessHandle handle) { + Matcher matcher = Pattern.compile("Error details saved to: (\\S+)") + .matcher(handle.getStderr()); + if (!matcher.find()) { + return ""; + } + Path errorLog = Path.of(matcher.group(1)); + try { + return "%n--- %s ---%n%s".formatted(errorLog, Files.readString(errorLog)); + } catch (IOException e) { + return "%n--- %s (unreadable: %s) ---".formatted(errorLog, e); } + } - @Override - public void write(byte[] b, int off, int len) throws IOException { - buffer.write(b, off, len); - console.write(b, off, len); + private static boolean containsAll(String output, String... expected) { + for (String part : expected) { + if (!output.contains(part)) { + return false; + } } + return true; + } - @Override - public void flush() throws IOException { - buffer.flush(); - console.flush(); + private static List cliLaunchCommand() { + List command = new ArrayList<>(); + if (isNative()) { + command.addAll(BuildExecutableOnce.getExecutablePath()); + command.addAll(forwardedSystemProperties()); + return command; + } + command.add(ProcessHandle.current().info().command().orElseThrow()); + command.addAll(tracingAgentArguments()); + command.addAll(forwardedSystemProperties()); + command.add("-jar"); + command.add(packagedCliJar().toString()); + return command; + } + + private static Path packagedCliJar() { + try (Stream files = Files.list(BUILD_OUTPUT_DIR)) { + return files + .filter(path -> path.getFileName().toString().endsWith("-runner.jar")) + .max(Comparator.comparing(CliBaseIT::lastModified)) + .orElseThrow(() -> new IllegalStateException( + "Packaged CLI (*-runner.jar) not found in target; run mvn package first")); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static FileTime lastModified(Path path) { + try { + return Files.getLastModifiedTime(path); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static List tracingAgentArguments() { + return ManagementFactory.getRuntimeMXBean().getInputArguments().stream() + .filter(argument -> argument.startsWith("-agentlib:native-image-agent")) + .map(argument -> argument.replace("config-merge-dir=", "config-output-dir=") + + "-async-" + ASYNC_COMMAND_COUNTER.incrementAndGet()) + .toList(); + } + + private static List forwardedSystemProperties() { + List arguments = new ArrayList<>(); + for (String name : System.getProperties().stringPropertyNames()) { + if (name.startsWith("streamx.") || name.startsWith("test.")) { + arguments.add("-D" + name + "=" + System.getProperty(name)); + } } + return arguments; } } diff --git a/src/test/java/com/streamx/cli/test/MeshTestSupport.java b/src/test/java/com/streamx/cli/test/MeshTestSupport.java index 2bd9816b..9c1012b3 100644 --- a/src/test/java/com/streamx/cli/test/MeshTestSupport.java +++ b/src/test/java/com/streamx/cli/test/MeshTestSupport.java @@ -11,7 +11,9 @@ import java.nio.file.Paths; import java.util.UUID; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -27,7 +29,15 @@ public final class MeshTestSupport { private static final Path PORT_RESERVATIONS = Paths.get(System.getProperty("java.io.tmpdir"), "streamx-cli-test-ports"); - public static final String CONTAINER_STARTUP_TIMEOUT_SECONDS = "300"; + /** Docker allocates its own published ports from 32768 upward, so we allocate below it. */ + private static final int PORT_RANGE_START = 20000; + + private static final int PORT_RANGE_SIZE = 12000; + + private static final AtomicInteger NEXT_PORT_OFFSET = + new AtomicInteger(ThreadLocalRandom.current().nextInt(PORT_RANGE_SIZE)); + + public static final String CONTAINER_STARTUP_TIMEOUT_SECONDS = "180"; private static volatile MeshManager activeMeshManager; private static volatile int activeProxyPort; @@ -44,18 +54,22 @@ private MeshTestSupport() { * The naive new ServerSocket(0) doesn't work reliably in this case. */ public static int freePort() { - for (int attempt = 0; attempt < 100; attempt++) { - int candidate; - try (ServerSocket s = new ServerSocket(0)) { - candidate = s.getLocalPort(); - } catch (IOException e) { - throw new RuntimeException("Failed to find a free port", e); - } - if (reserve(candidate)) { + for (int attempt = 0; attempt < PORT_RANGE_SIZE; attempt++) { + int candidate = PORT_RANGE_START + + Math.floorMod(NEXT_PORT_OFFSET.getAndIncrement(), PORT_RANGE_SIZE); + if (isFree(candidate) && reserve(candidate)) { return candidate; } } - throw new IllegalStateException("Could not reserve a free port after 100 attempts"); + throw new IllegalStateException("Could not reserve a free port"); + } + + private static boolean isFree(int port) { + try (ServerSocket probe = new ServerSocket(port)) { + return probe.getLocalPort() == port; + } catch (IOException alreadyInUse) { + return false; + } } private static boolean reserve(int port) { From 41b4c2abc4b3a7b27675aa70004adef4bbca3ed2 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:07:54 +0300 Subject: [PATCH 17/29] Force CI build From 3629d55bad98f79eee62281be79f6302b93b2641 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:07:57 +0300 Subject: [PATCH 18/29] Force CI build From a2a66a281c4841b15d4701084be2d79ee5dbd833 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:08:01 +0300 Subject: [PATCH 19/29] Force CI build From 511b8ec3b3baf57dd4c072edcc6395cee9fd78a6 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:08:04 +0300 Subject: [PATCH 20/29] Force CI build From 9111892f64ebd4b137c14e40f5453ee6b61e7f6d Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:08:07 +0300 Subject: [PATCH 21/29] Force CI build From 0945602e16a59dfed1063c2ee984e6189c3b77a6 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:08:10 +0300 Subject: [PATCH 22/29] Force CI build From db32b98866d8aa52417f51f1667de417dccb05a8 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:32:51 +0300 Subject: [PATCH 23/29] Force CI build From cd5c80ac147c72102b71a1ad12078f3658d4652d Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:32:54 +0300 Subject: [PATCH 24/29] Force CI build From 8cca5227ed65c475cb7ead280698cb579aac20ad Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:32:57 +0300 Subject: [PATCH 25/29] Force CI build From a7c4cd3d44b8486a24f72462f34292ac27a998e4 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:52:22 +0300 Subject: [PATCH 26/29] Force CI build From c9ffc6d3084b8c750d1114a1ff4538e84d42f806 Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:52:25 +0300 Subject: [PATCH 27/29] Force CI build From 4ade3aa26a78d74a788bf73f3747bcf8f25b2f4e Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:52:29 +0300 Subject: [PATCH 28/29] Force CI build From 6333dfe58f653edbc6885fdbe139bb14933007cb Mon Sep 17 00:00:00 2001 From: Kiryl Valkovich Date: Fri, 31 Jul 2026 12:52:32 +0300 Subject: [PATCH 29/29] Force CI build