This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 271e6a23c5e4 CAMEL-24511: camel-paho-mqtt5 - Restart route when
resubscribe fails after automatic reconnect
271e6a23c5e4 is described below
commit 271e6a23c5e47c32a1a458f7d0898d7767b08ec0
Author: JinyuChen97 <[email protected]>
AuthorDate: Thu Aug 27 16:22:58 2026 +0100
CAMEL-24511: camel-paho-mqtt5 - Restart route when resubscribe fails after
automatic reconnect
Backport of #25767 to camel-4.22.x.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
components/camel-paho-mqtt5/pom.xml | 5 +
.../component/paho/mqtt5/PahoMqtt5Consumer.java | 52 +++++-
.../mqtt5/PahoMqtt5ResubscribeFailureTest.java | 206 +++++++++++++++++++++
3 files changed, 258 insertions(+), 5 deletions(-)
diff --git a/components/camel-paho-mqtt5/pom.xml
b/components/camel-paho-mqtt5/pom.xml
index 1cbb8365baca..8c453ebe328f 100644
--- a/components/camel-paho-mqtt5/pom.xml
+++ b/components/camel-paho-mqtt5/pom.xml
@@ -52,6 +52,11 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- test infra -->
<dependency>
diff --git
a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
index ef6413031742..deec07b82392 100644
---
a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
+++
b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.paho.mqtt5;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+
import org.apache.camel.AsyncCallback;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
@@ -41,6 +44,7 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
private volatile String clientId;
private volatile boolean stopClient;
private volatile MqttConnectionOptions connectionOptions;
+ private final AtomicBoolean restarting = new AtomicBoolean(false);
public PahoMqtt5Consumer(Endpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -66,10 +70,7 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
clientId = PahoMqtt5Endpoint.generateClientId();
}
stopClient = true;
- client = new MqttClient(
- getEndpoint().getConfiguration().getBrokerUrl(),
- clientId,
-
PahoMqtt5Endpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
+ client = createClient();
LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
if (getEndpoint().getConfiguration().isManualAcksEnabled()) {
client.setManualAcks(true);
@@ -86,7 +87,16 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
try {
client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
} catch (MqttException e) {
- LOG.error("MQTT resubscribe failed {}",
e.getMessage(), e);
+ if (stopClient) {
+ LOG.warn("MQTT resubscribe failed on reconnect,
restarting route for recovery: {}",
+ e.getMessage(), e);
+ restartRouteAsync();
+ } else {
+ LOG.error(
+ "MQTT resubscribe failed on reconnect with
externally provided client,"
+ + " route will not be auto-restarted:
{}",
+ e.getMessage(), e);
+ }
}
}
}
@@ -126,6 +136,31 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
}
+ private void restartRouteAsync() {
+ if (!restarting.compareAndSet(false, true)) {
+ LOG.debug("Route restart already in progress, skipping duplicate
restart");
+ return;
+ }
+ String threadName = "PahoMqtt5-RestartRoute-" + getRouteId();
+ ExecutorService executor
+ =
getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
threadName);
+ executor.submit(() -> {
+ try {
+ String routeId = getRouteId();
+ LOG.info("Stopping route {} for restart after resubscribe
failure", routeId);
+
getEndpoint().getCamelContext().getRouteController().stopRoute(routeId);
+ LOG.info("Restarting route {}", routeId);
+
getEndpoint().getCamelContext().getRouteController().startRoute(routeId);
+ } catch (Exception e) {
+ getExceptionHandler().handleException(
+ "Failed to restart route after resubscribe failure",
e);
+ } finally {
+ restarting.set(false);
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ }
+ });
+ }
+
@Override
protected void doStop() throws Exception {
super.doStop();
@@ -145,6 +180,13 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
client = null;
}
+ MqttClient createClient() throws MqttException {
+ return new MqttClient(
+ getEndpoint().getConfiguration().getBrokerUrl(),
+ clientId,
+
PahoMqtt5Endpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
+ }
+
@Override
public PahoMqtt5Endpoint getEndpoint() {
return (PahoMqtt5Endpoint) super.getEndpoint();
diff --git
a/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ResubscribeFailureTest.java
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ResubscribeFailureTest.java
new file mode 100644
index 000000000000..5d952a9296a3
--- /dev/null
+++
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ResubscribeFailureTest.java
@@ -0,0 +1,206 @@
+/*
+ * 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.component.paho.mqtt5;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.eclipse.paho.mqttv5.client.MqttCallback;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PahoMqtt5ResubscribeFailureTest extends CamelTestSupport {
+
+ private static final String ROUTE_ID = "mqtt-consumer";
+
+ @Override
+ public boolean isUseAdviceWith() {
+ return true;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ }
+ };
+ }
+
+ private MqttCallback startRouteWithExternalClient(MqttClient mockClient)
throws Exception {
+ PahoMqtt5Endpoint endpoint = context.getEndpoint(
+ "paho-mqtt5:test?brokerUrl=tcp://localhost:1883",
PahoMqtt5Endpoint.class);
+ endpoint.setClient(mockClient);
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("paho-mqtt5:test?brokerUrl=tcp://localhost:1883")
+ .id(ROUTE_ID)
+ .to("mock:result");
+ }
+ });
+
+ context.start();
+
+ return captureCallback(mockClient);
+ }
+
+ private MqttCallback startRouteWithOwnedClient(MqttClient mockClient)
throws Exception {
+ PahoMqtt5Configuration config = new PahoMqtt5Configuration();
+ config.setBrokerUrl("tcp://localhost:1883");
+
+ PahoMqtt5Component component = new PahoMqtt5Component(context) {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ PahoMqtt5Endpoint endpoint = new PahoMqtt5Endpoint(uri,
remaining, this, config.copy()) {
+ @Override
+ public Consumer createConsumer(Processor processor) throws
Exception {
+ PahoMqtt5Consumer consumer = new
PahoMqtt5Consumer(this, processor) {
+ @Override
+ MqttClient createClient() {
+ return mockClient;
+ }
+ };
+ configureConsumer(consumer);
+ return consumer;
+ }
+ };
+ return endpoint;
+ }
+ };
+ context.addComponent("paho-mqtt5-owned", component);
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("paho-mqtt5-owned:test")
+ .id(ROUTE_ID)
+ .to("mock:result");
+ }
+ });
+
+ context.start();
+
+ return captureCallback(mockClient);
+ }
+
+ private MqttCallback captureCallback(MqttClient mockClient) throws
Exception {
+ ArgumentCaptor<MqttCallback> callbackCaptor =
ArgumentCaptor.forClass(MqttCallback.class);
+ verify(mockClient).setCallback(callbackCaptor.capture());
+ return callbackCaptor.getValue();
+ }
+
+ @Test
+ void resubscribeFailureWithExternalClientShouldNotRestartRoute() throws
Exception {
+ MqttClient mockClient = mock(MqttClient.class);
+ when(mockClient.isConnected()).thenReturn(true);
+
+ MqttCallback callback = startRouteWithExternalClient(mockClient);
+
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+
+ doThrow(new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION))
+ .when(mockClient).subscribe(anyString(), anyInt());
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+ await().during(2, TimeUnit.SECONDS)
+ .atMost(3, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID))
+ .isEqualTo(ServiceStatus.Started));
+ }
+
+ @Test
+ void resubscribeFailureWithOwnedClientShouldStopRoute() throws Exception {
+ MqttClient mockClient = mock(MqttClient.class);
+ when(mockClient.isConnected()).thenReturn(true);
+
+ MqttCallback callback = startRouteWithOwnedClient(mockClient);
+
+ doThrow(new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION))
+ .when(mockClient).subscribe(anyString(), anyInt());
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID))
+ .isEqualTo(ServiceStatus.Stopped));
+ }
+
+ @Test
+ void successfulResubscribeOnReconnectShouldKeepRouteStarted() throws
Exception {
+ MqttClient mockClient = mock(MqttClient.class);
+ when(mockClient.isConnected()).thenReturn(true);
+
+ MqttCallback callback = startRouteWithExternalClient(mockClient);
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+ verify(mockClient, times(2)).subscribe("test", 2);
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+ }
+
+ @Test
+ void duplicateReconnectsShouldNotCauseConcurrentRestarts() throws
Exception {
+ MqttClient mockClient = mock(MqttClient.class);
+ when(mockClient.isConnected()).thenReturn(true);
+
+ MqttCallback callback = startRouteWithOwnedClient(mockClient);
+
+ doThrow(new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION))
+ .when(mockClient).subscribe(anyString(), anyInt());
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID))
+ .isEqualTo(ServiceStatus.Stopped));
+ }
+
+ @Test
+ void initialConnectShouldNotResubscribe() throws Exception {
+ MqttClient mockClient = mock(MqttClient.class);
+ when(mockClient.isConnected()).thenReturn(true);
+
+ MqttCallback callback = startRouteWithExternalClient(mockClient);
+
+ callback.connectComplete(false, "tcp://localhost:1883");
+
+ verify(mockClient, times(1)).subscribe("test", 2);
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+ }
+}
