You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ja...@apache.org on 2026/09/17 15:26:41 UTC
(camel-quarkus) 01/02: Add integration tests for paho-mqtt5 resubscribe failure after reconnect
This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
commit 196816f206edb656f2f0ef62e205e55893334f30
Author: JinyuChen97 <as...@gmail.com>
AuthorDate: Wed Sep 2 14:19:09 2026 +0100
Add integration tests for paho-mqtt5 resubscribe failure after reconnect
Add two test scenarios using a custom FaultyMqtt5Broker (Java TCP server) that simulates MQTT5 resubscribe failure:
1. Recovery test: with SupervisingRouteController, route recovers via retry after restartRouteAsync() is triggered by keep-alive timeout
2. No-recovery test: without SupervisingRouteController, route correctly transitions to Stopped instead of remaining as a zombie in Started state
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
integration-tests/paho-mqtt5/pom.xml | 5 +
.../component/paho/mqtt5/it/PahoMqtt5Route.java | 19 ++-
.../component/paho/mqtt5/it/FaultyMqtt5Broker.java | 173 +++++++++++++++++++++
.../paho/mqtt5/it/InjectFaultyBroker.java | 27 ++++
.../mqtt5/it/PahoMqtt5ResubscribeNoRecoveryIT.java | 23 +++
.../it/PahoMqtt5ResubscribeNoRecoveryTest.java | 70 +++++++++
.../mqtt5/it/PahoMqtt5ResubscribeRecoveryIT.java | 23 +++
.../mqtt5/it/PahoMqtt5ResubscribeRecoveryTest.java | 75 +++++++++
.../paho/mqtt5/it/ResubscribeFailureProfile.java | 41 +++++
.../mqtt5/it/ResubscribeFailureTestResource.java | 55 +++++++
.../mqtt5/it/ResubscribeNoRecoveryProfile.java | 42 +++++
.../it/ResubscribeNoRecoveryTestResource.java | 55 +++++++
12 files changed, 603 insertions(+), 5 deletions(-)
diff --git a/integration-tests/paho-mqtt5/pom.xml b/integration-tests/paho-mqtt5/pom.xml
index d89a0c1e74..861fd7d42e 100644
--- a/integration-tests/paho-mqtt5/pom.xml
+++ b/integration-tests/paho-mqtt5/pom.xml
@@ -74,6 +74,11 @@
<artifactId>camel-quarkus-integration-tests-support-certificate-generator</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<profiles>
diff --git a/integration-tests/paho-mqtt5/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5Route.java b/integration-tests/paho-mqtt5/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5Route.java
index a9f0926217..172ce9f28e 100644
--- a/integration-tests/paho-mqtt5/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5Route.java
+++ b/integration-tests/paho-mqtt5/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5Route.java
@@ -34,12 +34,21 @@ public class PahoMqtt5Route extends RouteBuilder {
@Override
public void configure() throws Exception {
- SupervisingRouteController supervising = getCamelContext().getRouteController().supervising();
- supervising.setBackOffDelay(200);
- supervising.setIncludeRoutes("paho-mqtt5:*");
+ boolean enableSupervising = ConfigProvider.getConfig()
+ .getOptionalValue("paho5.route-controller.supervising", Boolean.class)
+ .orElse(true);
+ if (enableSupervising) {
+ SupervisingRouteController supervising = getCamelContext().getRouteController().supervising();
+ supervising.setBackOffDelay(200);
+ supervising.setIncludeRoutes("paho-mqtt5:*");
+ }
- from("direct:test").to("paho-mqtt5:queue?lazyStartProducer=true&brokerUrl=" + brokerUrl("tcp"));
- from("paho-mqtt5:queue?brokerUrl=" + brokerUrl("tcp"))
+ String keepAliveOpt = ConfigProvider.getConfig()
+ .getOptionalValue("paho5.keep-alive-interval", String.class)
+ .map(v -> "&keepAliveInterval=" + v)
+ .orElse("");
+ from("direct:test").to("paho-mqtt5:queue?lazyStartProducer=true&brokerUrl=" + brokerUrl("tcp") + keepAliveOpt);
+ from("paho-mqtt5:queue?brokerUrl=" + brokerUrl("tcp") + keepAliveOpt)
.id(TESTING_ROUTE_ID)
.routePolicy(new RoutePolicySupport() {
@Override
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/FaultyMqtt5Broker.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/FaultyMqtt5Broker.java
new file mode 100644
index 0000000000..0b970b067e
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/FaultyMqtt5Broker.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Minimal MQTT 5 TCP server that reproduces the resubscribe-failure scenario from CAMEL-24511.
+ *
+ * <ul>
+ * <li>Connection 1 (initial): CONNECT/CONNACK, SUBSCRIBE/SUBACK, then force-close to simulate connection
+ * loss.</li>
+ * <li>Connection 2 (auto-reconnect): CONNECT/CONNACK, but ignores SUBSCRIBE and PINGREQ. The client's
+ * keep-alive timer fires, causing MqttException 32000 (connection lost) which triggers the fix.</li>
+ * <li>Connection 3+ (after route restart): normal CONNECT/CONNACK, SUBSCRIBE/SUBACK, PINGREQ/PINGRESP.</li>
+ * </ul>
+ */
+class FaultyMqtt5Broker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FaultyMqtt5Broker.class);
+
+ private final int port;
+ private final int firstNormalConnection;
+ private final AtomicInteger connectionCount = new AtomicInteger(0);
+ private volatile ServerSocket serverSocket;
+ private volatile ExecutorService executor;
+ private volatile boolean running;
+
+ FaultyMqtt5Broker(int port, int firstNormalConnection) {
+ this.port = port;
+ this.firstNormalConnection = firstNormalConnection;
+ }
+
+ void start() throws IOException {
+ serverSocket = new ServerSocket(port);
+ running = true;
+ executor = Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r, "FaultyMqtt5Broker");
+ t.setDaemon(true);
+ return t;
+ });
+ executor.submit(this::acceptLoop);
+ LOG.info("FaultyMqtt5Broker started on port {}", port);
+ }
+
+ void stop() {
+ running = false;
+ try {
+ if (serverSocket != null) {
+ serverSocket.close();
+ }
+ } catch (IOException e) {
+ // ignored
+ }
+ if (executor != null) {
+ executor.shutdownNow();
+ }
+ LOG.info("FaultyMqtt5Broker stopped");
+ }
+
+ int getConnectionCount() {
+ return connectionCount.get();
+ }
+
+ private void acceptLoop() {
+ while (running) {
+ try {
+ Socket socket = serverSocket.accept();
+ int connNum = connectionCount.incrementAndGet();
+ LOG.info("Connection #{} from {}", connNum, socket.getRemoteSocketAddress());
+ executor.submit(() -> handleConnection(socket, connNum));
+ } catch (SocketException e) {
+ if (running) {
+ LOG.warn("Accept error: {}", e.getMessage());
+ }
+ } catch (IOException e) {
+ LOG.warn("Accept error: {}", e.getMessage());
+ }
+ }
+ }
+
+ private void handleConnection(Socket socket, int connNum) {
+ try (socket) {
+ InputStream in = socket.getInputStream();
+ OutputStream out = socket.getOutputStream();
+
+ byte[] buf = new byte[1024];
+ int len = in.read(buf);
+ if (len <= 0 || ((buf[0] >> 4) & 0x0F) != 1) {
+ return;
+ }
+ // CONNACK: session not present, success, no properties
+ LOG.info("Conn #{}: CONNECT received, sending CONNACK", connNum);
+ out.write(new byte[] { 0x20, 0x03, 0x00, 0x00, 0x00 });
+ out.flush();
+
+ if (connNum >= 2 && connNum < firstNormalConnection) {
+ LOG.info("Conn #{}: Simulating resubscribe failure - not responding to SUBSCRIBE or PINGREQ", connNum);
+ // Keep socket open but ignore everything.
+ // The client's keep-alive timer will fire, causing MqttException 32000.
+ try {
+ while (running && in.read(buf) > 0) {
+ int pt = (buf[0] >> 4) & 0x0F;
+ LOG.debug("Conn #{}: Ignoring packet type {}", connNum, pt);
+ }
+ } catch (IOException ignored) {
+ }
+ return;
+ }
+
+ // Connection 1 and 3+: handle packets normally
+ while (running) {
+ len = in.read(buf);
+ if (len <= 0) {
+ break;
+ }
+ int packetType = (buf[0] >> 4) & 0x0F;
+ switch (packetType) {
+ case 8: // SUBSCRIBE
+ int packetIdMsb = len > 2 ? buf[2] : 0;
+ int packetIdLsb = len > 3 ? buf[3] : 1;
+ LOG.info("Conn #{}: SUBSCRIBE received, sending SUBACK", connNum);
+ out.write(new byte[] { (byte) 0x90, 0x04, (byte) packetIdMsb, (byte) packetIdLsb, 0x00, 0x01 });
+ out.flush();
+ if (connNum == 1) {
+ Thread.sleep(2000);
+ LOG.info("Conn #{}: Simulating connection loss - closing connection", connNum);
+ return;
+ }
+ break;
+ case 12: // PINGREQ
+ out.write(new byte[] { (byte) 0xD0, 0x00 });
+ out.flush();
+ break;
+ case 14: // DISCONNECT
+ LOG.info("Conn #{}: DISCONNECT received", connNum);
+ return;
+ default:
+ break;
+ }
+ }
+ } catch (SocketException e) {
+ LOG.debug("Conn #{}: Socket closed: {}", connNum, e.getMessage());
+ } catch (Exception e) {
+ LOG.warn("Conn #{}: Error: {}", connNum, e.getMessage(), e);
+ }
+ }
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/InjectFaultyBroker.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/InjectFaultyBroker.java
new file mode 100644
index 0000000000..f8d7146d2e
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/InjectFaultyBroker.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ ElementType.ANNOTATION_TYPE, ElementType.METHOD, ElementType.FIELD })
+@Retention(RetentionPolicy.RUNTIME)
+public @interface InjectFaultyBroker {
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeNoRecoveryIT.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeNoRecoveryIT.java
new file mode 100644
index 0000000000..16b8f21807
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeNoRecoveryIT.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+
+@QuarkusIntegrationTest
+class PahoMqtt5ResubscribeNoRecoveryIT extends PahoMqtt5ResubscribeNoRecoveryTest {
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeNoRecoveryTest.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeNoRecoveryTest.java
new file mode 100644
index 0000000000..002a72dc7b
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeNoRecoveryTest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.restassured.RestAssured;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.is;
+
+/**
+ * Integration test for CAMEL-24511: verifies that without {@code SupervisingRouteController}, a paho-mqtt5 consumer
+ * route transitions to Stopped (not zombie Started) after a resubscribe failure on automatic reconnect.
+ *
+ * The broker never sends SUBACK after the initial connection, so the route cannot recover. Without
+ * {@code SupervisingRouteController}, there is no retry mechanism — the route stays Stopped.
+ */
+@QuarkusTest
+@TestProfile(ResubscribeNoRecoveryProfile.class)
+class PahoMqtt5ResubscribeNoRecoveryTest {
+
+ @InjectFaultyBroker
+ FaultyMqtt5Broker broker;
+
+ @Test
+ void resubscribeFailureWithoutSupervisingControllerShouldStopRoute() {
+ // 1. Wait for the route to be Started (connection 1 to faulty broker succeeds)
+ await().atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(() -> RestAssured.get("/paho-mqtt5/routeStatus/" + PahoMqtt5Route.TESTING_ROUTE_ID)
+ .then()
+ .statusCode(200)
+ .body(is("Started")));
+
+ // 2. The faulty broker closes connection 1 after ~2s (simulating connection loss),
+ // Paho auto-reconnects (connection 2),
+ // but the broker never sends SUBACK — keep-alive timeout fires (~7.5s with keepAlive=5),
+ // our fix calls restartRouteAsync(), route stops. Without SupervisingRouteController,
+ // startRoute() also fails (broker still not sending SUBACK) and the route stays Stopped.
+ await().atMost(60, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until(() -> broker.getConnectionCount() >= 2);
+
+ // 3. The route should transition to Stopped — NOT remain as a zombie in Started state
+ await().atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(() -> RestAssured.get("/paho-mqtt5/routeStatus/" + PahoMqtt5Route.TESTING_ROUTE_ID)
+ .then()
+ .statusCode(200)
+ .body(is("Stopped")));
+ }
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeRecoveryIT.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeRecoveryIT.java
new file mode 100644
index 0000000000..1390def85c
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeRecoveryIT.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+
+@QuarkusIntegrationTest
+class PahoMqtt5ResubscribeRecoveryIT extends PahoMqtt5ResubscribeRecoveryTest {
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeRecoveryTest.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeRecoveryTest.java
new file mode 100644
index 0000000000..2b26b36b1f
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5ResubscribeRecoveryTest.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.restassured.RestAssured;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.is;
+
+/**
+ * Integration test for CAMEL-24511: verifies that a paho-mqtt5 consumer route recovers after a resubscribe failure on
+ * automatic reconnect, instead of silently becoming a zombie.
+ *
+ * Uses {@link FaultyMqtt5Broker} (a Java TCP server) to simulate a broker that accepts the initial connection normally
+ * but fails to send SUBACK on connections 2 and 3 (auto-reconnect and first restart attempt), triggering keep-alive
+ * timeouts. The fix in {@code PahoMqtt5Consumer.restartRouteAsync()} detects this and restarts the route. The first
+ * restart fails (connection 3), but {@code SupervisingRouteController} retries with exponential backoff, and the route
+ * recovers on connection 4 when the broker starts responding normally again.
+ */
+@QuarkusTest
+@TestProfile(ResubscribeFailureProfile.class)
+class PahoMqtt5ResubscribeRecoveryTest {
+
+ @InjectFaultyBroker
+ FaultyMqtt5Broker broker;
+
+ @Test
+ void resubscribeFailureShouldTriggerRouteRestartAndRecovery() {
+ // 1. Wait for the route to be Started (connection 1 to faulty broker succeeds)
+ await().atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(() -> RestAssured.get("/paho-mqtt5/routeStatus/" + PahoMqtt5Route.TESTING_ROUTE_ID)
+ .then()
+ .statusCode(200)
+ .body(is("Started")));
+
+ // 2. The faulty broker closes connection 1 after ~2s (simulating connection loss),
+ // Paho auto-reconnects (connection 2),
+ // but the broker ignores SUBSCRIBE — keep-alive timeout fires (~7.5s with keepAlive=5),
+ // our fix calls restartRouteAsync(), route stops. The restart attempt (connection 3)
+ // also fails (broker still ignores SUBSCRIBE). SupervisingRouteController retries with
+ // backoff, and connection 4 succeeds (broker behaves normally from connection 4 onward).
+ // Wait for at least 4 connections — proves the SupervisingRouteController retry happened.
+ await().atMost(90, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until(() -> broker.getConnectionCount() >= 4);
+
+ // 3. After SupervisingRouteController retries, the route should recover to Started
+ await().atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(() -> RestAssured.get("/paho-mqtt5/routeStatus/" + PahoMqtt5Route.TESTING_ROUTE_ID)
+ .then()
+ .statusCode(200)
+ .body(is("Started")));
+ }
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeFailureProfile.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeFailureProfile.java
new file mode 100644
index 0000000000..26186fa2f0
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeFailureProfile.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.util.List;
+import java.util.Map;
+
+import io.quarkus.test.junit.QuarkusTestProfile;
+
+public class ResubscribeFailureProfile implements QuarkusTestProfile {
+
+ @Override
+ public List<TestResourceEntry> testResources() {
+ return List.of(new TestResourceEntry(ResubscribeFailureTestResource.class));
+ }
+
+ @Override
+ public boolean disableGlobalTestResources() {
+ return true;
+ }
+
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return Map.of(
+ "paho5.keep-alive-interval", "5");
+ }
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeFailureTestResource.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeFailureTestResource.java
new file mode 100644
index 0000000000..abe05bab10
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeFailureTestResource.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.util.Map;
+
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+import org.apache.camel.quarkus.test.AvailablePortFinder;
+import org.apache.camel.util.CollectionHelper;
+
+public class ResubscribeFailureTestResource implements QuarkusTestResourceLifecycleManager {
+
+ private FaultyMqtt5Broker broker;
+ private int port;
+
+ @Override
+ public Map<String, String> start() {
+ port = AvailablePortFinder.getNextAvailable();
+ broker = new FaultyMqtt5Broker(port, 4);
+ try {
+ broker.start();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to start FaultyMqtt5Broker", e);
+ }
+ return CollectionHelper.mapOf(
+ "paho5.broker.tcp.url", "tcp://localhost:" + port);
+ }
+
+ @Override
+ public void inject(TestInjector testInjector) {
+ testInjector.injectIntoFields(broker,
+ new TestInjector.AnnotatedAndMatchesType(InjectFaultyBroker.class, FaultyMqtt5Broker.class));
+ }
+
+ @Override
+ public void stop() {
+ if (broker != null) {
+ broker.stop();
+ }
+ }
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeNoRecoveryProfile.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeNoRecoveryProfile.java
new file mode 100644
index 0000000000..af092721c6
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeNoRecoveryProfile.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.util.List;
+import java.util.Map;
+
+import io.quarkus.test.junit.QuarkusTestProfile;
+
+public class ResubscribeNoRecoveryProfile implements QuarkusTestProfile {
+
+ @Override
+ public List<TestResourceEntry> testResources() {
+ return List.of(new TestResourceEntry(ResubscribeNoRecoveryTestResource.class));
+ }
+
+ @Override
+ public boolean disableGlobalTestResources() {
+ return true;
+ }
+
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return Map.of(
+ "paho5.keep-alive-interval", "5",
+ "paho5.route-controller.supervising", "false");
+ }
+}
diff --git a/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeNoRecoveryTestResource.java b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeNoRecoveryTestResource.java
new file mode 100644
index 0000000000..8c1ac4a5ac
--- /dev/null
+++ b/integration-tests/paho-mqtt5/src/test/java/org/apache/camel/quarkus/component/paho/mqtt5/it/ResubscribeNoRecoveryTestResource.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.paho.mqtt5.it;
+
+import java.util.Map;
+
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+import org.apache.camel.quarkus.test.AvailablePortFinder;
+import org.apache.camel.util.CollectionHelper;
+
+public class ResubscribeNoRecoveryTestResource implements QuarkusTestResourceLifecycleManager {
+
+ private FaultyMqtt5Broker broker;
+ private int port;
+
+ @Override
+ public Map<String, String> start() {
+ port = AvailablePortFinder.getNextAvailable();
+ broker = new FaultyMqtt5Broker(port, Integer.MAX_VALUE);
+ try {
+ broker.start();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to start FaultyMqtt5Broker", e);
+ }
+ return CollectionHelper.mapOf(
+ "paho5.broker.tcp.url", "tcp://localhost:" + port);
+ }
+
+ @Override
+ public void inject(TestInjector testInjector) {
+ testInjector.injectIntoFields(broker,
+ new TestInjector.AnnotatedAndMatchesType(InjectFaultyBroker.class, FaultyMqtt5Broker.class));
+ }
+
+ @Override
+ public void stop() {
+ if (broker != null) {
+ broker.stop();
+ }
+ }
+}