This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 443ebb407373 CAMEL-24511: camel-paho-mqtt5 - restart route on
resubscribe failure
443ebb407373 is described below
commit 443ebb4073734e4fe26c5f9fe3bf072d063f9a74
Author: JinyuChen97 <[email protected]>
AuthorDate: Thu Aug 27 15:46:04 2026 +0100
CAMEL-24511: camel-paho-mqtt5 - restart route on resubscribe failure
When automaticReconnect is enabled and the broker reconnects, if the
post-reconnect subscribe() fails (e.g. SUBACK never arrives and the Paho
keepAlive timer triggers MqttException 32000), the consumer now stops and
restarts the route instead of only logging the error. This prevents the
route from silently entering a zombie state where it shows Started but
consumes no messages. If the restart also fails, the route is left
Stopped, so a SupervisingRouteController can keep retrying with
exponential backoff until the broker recovers.
Closes #25767
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../component/paho/mqtt5/PahoMqtt5Consumer.java | 40 +++-
.../mqtt5/PahoMqtt5ResubscribeFailureTest.java | 206 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 21 +++
3 files changed, 266 insertions(+), 1 deletion(-)
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 0f1b5faf661a..bfe641e1ba95 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);
@@ -83,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);
+ }
}
}
}
@@ -139,6 +152,31 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
}
}
+ 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 {
MqttClient ownedClient = stopClient ? client : null;
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);
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 4eb6703b0e49..b3c9bc1e2826 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -456,3 +456,24 @@ knative:endpoint/myEndpoint?muteException=false
`org.apache.camel.component.knative.spi.KnativeTransportConfiguration` gains a
fourth constructor
argument for the flag. The three-argument constructor is retained and mutes
the exception, so existing
code compiles unchanged and picks up the new default.
+
+=== camel-paho-mqtt5
+
+When `automaticReconnect=true` and the MQTT broker reconnects, the consumer
now restarts the route
+if the post-reconnect `subscribe()` call fails. Previously a failed
resubscription (for example,
+when the broker does not send a SUBACK and the Paho keepAlive timer triggers
`MqttException 32000`)
+was only logged at ERROR level with no recovery action, leaving the route in
`Started` state while
+silently consuming no messages (zombie state).
+
+If the resubscribe fails and the consumer owns the MQTT client (the default),
it automatically stops
+and restarts the route to force a clean reconnect. If the restart also fails
(for example, the broker
+is still unavailable), the route is left in `Stopped` state. Routes using a
user-provided client are
+not affected by this change. Configuring Camel's `SupervisingRouteController`
allows the framework to
+keep retrying with exponential backoff until the broker recovers:
+
+[source,properties]
+----
+camel.routeController.enabled = true
+camel.routeController.backOffDelay = 2000
+camel.routeController.backOffMaxDelay = 60000
+----
