From 2860fdae87f7afdca5824497b4ddc2da75a8ed80 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:44:30 +0000 Subject: [PATCH 1/3] Build(deps): Bump org.codehaus.woodstox:stax2-api from 4.3.0 to 4.3.1 Bumps [org.codehaus.woodstox:stax2-api](https://github.com/FasterXML/stax2-api) from 4.3.0 to 4.3.1. - [Commits](https://github.com/FasterXML/stax2-api/compare/stax2-api-4.3.0...stax2-api-4.3.1) --- updated-dependencies: - dependency-name: org.codehaus.woodstox:stax2-api dependency-version: 4.3.1 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] --- .../basyx.digitaltwinregistry.component/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/basyx.aasdigitaltwinregistry/basyx.digitaltwinregistry.component/pom.xml b/basyx.aasdigitaltwinregistry/basyx.digitaltwinregistry.component/pom.xml index bad65fce2..3aace73b2 100644 --- a/basyx.aasdigitaltwinregistry/basyx.digitaltwinregistry.component/pom.xml +++ b/basyx.aasdigitaltwinregistry/basyx.digitaltwinregistry.component/pom.xml @@ -179,7 +179,7 @@ org.codehaus.woodstox stax2-api - 4.3.0 + 4.3.1 From 84e8bfb00b53165f6e3cc4a45b62d927ba2448ac Mon Sep 17 00:00:00 2001 From: Aaron Zielstorff Date: Tue, 6 Oct 2026 09:46:04 +0200 Subject: [PATCH 2/3] Harden MQTT broker test fixture against Paho in-flight race Paho completes a synchronous QoS 1 publish before releasing its in-flight slot on the callback thread. On slow CI runners these slots accumulate and exceed the default window of 10, failing publishesConcurrentlyFromMultipleClients with 'Too many publishes in progress (32202)'. Connect fixture clients with a larger in-flight window. Signed-off-by: Aaron Zielstorff --- .../common/mqttcore/MqttBrokerTestSupport.java | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java b/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java index 401e8bc8d..82ab6f6cb 100644 --- a/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java +++ b/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java @@ -36,6 +36,7 @@ import org.eclipse.digitaltwin.basyx.common.mqttcore.listener.MqttTestListener; import org.eclipse.paho.client.mqttv3.IMqttClient; import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import io.moquette.broker.Server; @@ -48,6 +49,15 @@ public final class MqttBrokerTestSupport implements AutoCloseable { private static final Duration PORT_BIND_TIMEOUT = Duration.ofSeconds(10); private static final String WEBSOCKET_TRANSPORT_NAME = "Websocket MQTT"; + /** + * Paho wakes a synchronous QoS 1 publish when the PUBACK arrives, but only + * releases the in-flight slot later on its callback thread. On slow machines + * these unreleased slots pile up and exceed Paho's default window of 10, which + * fails the next publish with "Too many publishes in progress (32202)". The + * window is therefore sized above the number of messages any test publishes + * through a single client. + */ + private static final int TEST_CLIENT_MAX_INFLIGHT = 1000; private final MqttTestListener listener = new MqttTestListener(); private final Server broker; private final List clients = new ArrayList<>(); @@ -86,7 +96,9 @@ public static MqttBrokerTestSupport startWithWebSocket() throws IOException { public MqttClient connectClient() throws MqttException { MqttClient client = new MqttClient(serverUri(), uniqueClientId()); clients.add(client); - client.connect(); + MqttConnectOptions options = new MqttConnectOptions(); + options.setMaxInflight(TEST_CLIENT_MAX_INFLIGHT); + client.connect(options); return client; } From 8e4994bcca992b4e0719b9f336e19f835687630a Mon Sep 17 00:00:00 2001 From: Aaron Zielstorff Date: Tue, 6 Oct 2026 10:02:04 +0200 Subject: [PATCH 3/3] Assign explicit ports in MQTT broker test fixture Moquette 0.17 stores bound ephemeral ports in a plain HashMap written from a Netty thread, while Server#getPort() inserts a 0 placeholder into the same map from the caller thread. Under contention the real port is lost and the fixture timed out with 'Timed out awaiting Moquette TCP port'. Reserve free ports up front, pass them to Moquette explicitly and retry on bind conflicts. This also removes the reflective WebSocket port lookup. Signed-off-by: Aaron Zielstorff --- .../mqttcore/MqttBrokerTestSupport.java | 135 ++++++++++-------- .../mqttcore/MqttClientConfigurationTest.java | 2 +- 2 files changed, 78 insertions(+), 59 deletions(-) diff --git a/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java b/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java index 82ab6f6cb..190d34506 100644 --- a/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java +++ b/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttBrokerTestSupport.java @@ -25,13 +25,12 @@ package org.eclipse.digitaltwin.basyx.common.mqttcore; import java.io.IOException; -import java.lang.reflect.Field; -import java.time.Duration; +import java.net.BindException; +import java.net.InetAddress; +import java.net.ServerSocket; import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.UUID; -import java.util.function.IntSupplier; import org.eclipse.digitaltwin.basyx.common.mqttcore.listener.MqttTestListener; import org.eclipse.paho.client.mqttv3.IMqttClient; @@ -47,8 +46,8 @@ * Isolated, in-memory Moquette fixture shared by MQTT feature tests. */ public final class MqttBrokerTestSupport implements AutoCloseable { - private static final Duration PORT_BIND_TIMEOUT = Duration.ofSeconds(10); - private static final String WEBSOCKET_TRANSPORT_NAME = "Websocket MQTT"; + private static final String BROKER_HOST = "localhost"; + private static final int MAX_START_ATTEMPTS = 5; /** * Paho wakes a synchronous QoS 1 publish when the PUBACK arrives, but only * releases the in-flight slot later on its callback thread. On slow machines @@ -58,39 +57,91 @@ public final class MqttBrokerTestSupport implements AutoCloseable { * through a single client. */ private static final int TEST_CLIENT_MAX_INFLIGHT = 1000; - private final MqttTestListener listener = new MqttTestListener(); + private final MqttTestListener listener; private final Server broker; private final List clients = new ArrayList<>(); private final int port; private final int websocketPort; - private MqttBrokerTestSupport(IConfig config, boolean websocketEnabled) throws IOException { - broker = new Server(); - try { - broker.startServer(config, List.of(listener)); - port = awaitBoundPort(broker::getPort, "TCP"); - websocketPort = websocketEnabled ? awaitBoundPort(() -> readTransportPort(broker, WEBSOCKET_TRANSPORT_NAME), "WebSocket") : -1; - } catch (IOException | RuntimeException | Error startupError) { - try { - broker.stopServer(); - } catch (Throwable cleanupError) { - startupError.addSuppressed(cleanupError); - } - throw startupError; - } + private MqttBrokerTestSupport(Server broker, MqttTestListener listener, int port, int websocketPort) { + this.broker = broker; + this.listener = listener; + this.port = port; + this.websocketPort = websocketPort; } public static MqttBrokerTestSupport start() throws IOException { - return start(new FluentConfig().host("localhost").port(0).allowAnonymous().disablePersistence().disableTelemetry().build()); + return start(new FluentConfig().host(BROKER_HOST).allowAnonymous().disablePersistence().disableTelemetry().build()); } + /** + * Starts a broker with the given configuration. The TCP port is always + * assigned by the fixture and overrides any port in {@code config}. + */ public static MqttBrokerTestSupport start(IConfig config) throws IOException { - return new MqttBrokerTestSupport(config, false); + return startOnFreePorts(config, false); } public static MqttBrokerTestSupport startWithWebSocket() throws IOException { - IConfig config = new FluentConfig().host("localhost").port(0).websocketPort(0).allowAnonymous().disablePersistence().disableTelemetry().build(); - return new MqttBrokerTestSupport(config, true); + return startOnFreePorts(new FluentConfig().host(BROKER_HOST).allowAnonymous().disablePersistence().disableTelemetry().build(), true); + } + + /** + * Moquette 0.17 cannot reliably report ephemeral ports: the acceptor stores + * them in a plain HashMap from a Netty thread, while Server#getPort() writes a + * 0 placeholder into the same map from the caller thread. Under contention the + * real port is lost and getPort() keeps returning 0. The fixture therefore + * reserves concrete ports up front and retries if another process grabs one + * before Moquette binds it. + */ + private static MqttBrokerTestSupport startOnFreePorts(IConfig config, boolean websocketEnabled) throws IOException { + RuntimeException lastBindFailure = null; + for (int attempt = 0; attempt < MAX_START_ATTEMPTS; attempt++) { + int port = reserveFreePort(); + int websocketPort = websocketEnabled ? reserveFreePort() : -1; + if (port == websocketPort) { + continue; + } + config.setProperty(IConfig.PORT_PROPERTY_NAME, Integer.toString(port)); + if (websocketEnabled) { + config.setProperty(IConfig.WEB_SOCKET_PORT_PROPERTY_NAME, Integer.toString(websocketPort)); + } + MqttTestListener listener = new MqttTestListener(); + Server broker = new Server(); + try { + broker.startServer(config, List.of(listener)); + return new MqttBrokerTestSupport(broker, listener, port, websocketPort); + } catch (IOException | RuntimeException | Error startupError) { + try { + broker.stopServer(); + } catch (Throwable cleanupError) { + startupError.addSuppressed(cleanupError); + } + if (!(startupError instanceof RuntimeException runtimeError) || !isBindFailure(runtimeError)) { + throw startupError; + } + if (lastBindFailure != null) { + runtimeError.addSuppressed(lastBindFailure); + } + lastBindFailure = runtimeError; + } + } + throw new IOException("Could not bind Moquette to a free port after " + MAX_START_ATTEMPTS + " attempts", lastBindFailure); + } + + private static int reserveFreePort() throws IOException { + try (ServerSocket socket = new ServerSocket(0, 0, InetAddress.getByName(BROKER_HOST))) { + return socket.getLocalPort(); + } + } + + private static boolean isBindFailure(Throwable error) { + for (Throwable cause = error; cause != null; cause = cause.getCause()) { + if (cause instanceof BindException) { + return true; + } + } + return false; } public MqttClient connectClient() throws MqttException { @@ -108,7 +159,7 @@ public T trackClient(T client) { } public String serverUri() { - return "tcp://localhost:" + port; + return "tcp://" + BROKER_HOST + ":" + port; } public int port() { @@ -130,38 +181,6 @@ private static String uniqueClientId() { return "mqtt-" + UUID.randomUUID().toString().substring(0, 16); } - private static int awaitBoundPort(IntSupplier portSupplier, String transport) throws IOException { - long deadline = System.nanoTime() + PORT_BIND_TIMEOUT.toNanos(); - do { - int boundPort = portSupplier.getAsInt(); - if (boundPort > 0) { - return boundPort; - } - try { - Thread.sleep(10); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IOException("Interrupted while awaiting Moquette " + transport + " port", e); - } - } while (System.nanoTime() < deadline); - throw new IOException("Timed out awaiting Moquette " + transport + " port"); - } - - @SuppressWarnings("unchecked") - private static int readTransportPort(Server broker, String transportName) { - try { - Field acceptorField = Server.class.getDeclaredField("acceptor"); - acceptorField.setAccessible(true); - Object acceptor = acceptorField.get(broker); - Field portsField = acceptor.getClass().getDeclaredField("ports"); - portsField.setAccessible(true); - Map ports = (Map) portsField.get(acceptor); - return ports.getOrDefault(transportName, -1); - } catch (ReflectiveOperationException e) { - throw new IllegalStateException("Could not read Moquette's bound " + transportName + " port", e); - } - } - @Override public void close() throws MqttException { Throwable failure = null; diff --git a/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttClientConfigurationTest.java b/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttClientConfigurationTest.java index 15289d8b6..1481d4d39 100644 --- a/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttClientConfigurationTest.java +++ b/basyx.common/basyx.mqttcore/src/test/java/org/eclipse/digitaltwin/basyx/common/mqttcore/MqttClientConfigurationTest.java @@ -65,7 +65,7 @@ public void connectsAnonymouslyOverTcp() throws Exception { public void acceptsValidAndRejectsInvalidTcpCredentials() throws Exception { Path passwordFile = Files.createTempFile("moquette-passwords-", ".conf"); Files.writeString(passwordFile, USERNAME + ":" + PASSWORD_HASH + System.lineSeparator(), StandardCharsets.UTF_8); - try (MqttBrokerTestSupport fixture = MqttBrokerTestSupport.start(new FluentConfig().host("localhost").port(0).disallowAnonymous() + try (MqttBrokerTestSupport fixture = MqttBrokerTestSupport.start(new FluentConfig().host("localhost").disallowAnonymous() .passwordFile(passwordFile.toAbsolutePath().toString()).disablePersistence().disableTelemetry().build())) { IMqttClient validClient = fixture.trackClient( configuration.mqttClient(uniqueClientId(), "localhost", fixture.port(), "tcp", configuration.mqttConnectOptions(USERNAME, PASSWORD)));