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 737ed0f5b35b CAMEL-24523: camel-paho - restart the consumer route when
resubscribe fails after reconnect
737ed0f5b35b is described below
commit 737ed0f5b35b349632ad9632403b72f60a406ddb
Author: nkokitkar <[email protected]>
AuthorDate: Wed Sep 16 01:48:58 2026 -0700
CAMEL-24523: camel-paho - restart the consumer route when resubscribe fails
after reconnect
When Paho's automatic reconnect succeeds but the topic resubscription
fails, the consumer silently stops receiving messages while the route
still reports Started. The consumer now restarts its route asynchronously
(off the Paho callback thread, on a dedicated executor, a CAS against a
double restart, stale callbacks of a replaced client ignored, no restart
while the consumer or the context is stopping) when it owns the client;
with an externally provided client it logs an error and leaves the route
alone, keeping the owned/shared cleanup semantics of CAMEL-24465.
Mirrors the camel-paho-mqtt5 recovery of CAMEL-24511. camel-paho is
deprecated; this is a narrow reliability fix for MQTT 3.1/3.1.1 users
who cannot migrate yet. Documented in the 4.23 upgrade guide with the
cleanSession and SupervisingRouteController guidance.
Closes #26476
Co-authored-by: Copilot <[email protected]>
---
.../apache/camel/component/paho/PahoConsumer.java | 69 +++++-
.../component/paho/PahoResubscribeFailureTest.java | 250 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 22 ++
3 files changed, 336 insertions(+), 5 deletions(-)
diff --git
a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
index 19269c2cc05e..2727a81418f1 100644
---
a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
+++
b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.paho;
+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;
@@ -40,6 +43,7 @@ public class PahoConsumer extends DefaultConsumer {
private volatile String clientId;
private volatile boolean stopClient;
private volatile MqttConnectOptions connectOptions;
+ private final AtomicBoolean restarting = new AtomicBoolean(false);
public PahoConsumer(Endpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -74,15 +78,26 @@ public class PahoConsumer extends DefaultConsumer {
client.connect(connectOptions);
}
- client.setCallback(new MqttCallbackExtended() {
+ MqttClient callbackClient = client;
+ boolean isOwnedClient = stopClient;
+ callbackClient.setCallback(new MqttCallbackExtended() {
@Override
public void connectComplete(boolean reconnect, String
serverURI) {
- if (reconnect) {
+ if (reconnect && isRunAllowed() && callbackClient ==
client) {
try {
- client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ callbackClient.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
} catch (MqttException e) {
- LOG.error("MQTT resubscribe failed {}",
e.getMessage(), e);
+ if (isOwnedClient) {
+ 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);
+ }
}
}
}
@@ -109,7 +124,7 @@ public class PahoConsumer extends DefaultConsumer {
});
LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
- client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ callbackClient.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
} catch (Exception startException) {
MqttClient ownedClient = stopClient ? client : null;
if (ownedClient != null) {
@@ -128,6 +143,50 @@ public class PahoConsumer extends DefaultConsumer {
}
}
+ private void restartRouteAsync() {
+ if (!restarting.compareAndSet(false, true)) {
+ LOG.debug("Route restart already in progress, skipping duplicate
restart");
+ return;
+ }
+ String threadName = "Paho-RestartRoute-" + getRouteId();
+ ExecutorService executor = null;
+ try {
+ executor
+ =
getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
threadName);
+ ExecutorService restartExecutor = executor;
+ restartExecutor.submit(() -> {
+ try {
+ if (!isRunAllowed() ||
!getEndpoint().getCamelContext().isRunAllowed()) {
+ LOG.debug("Consumer or Camel context is stopping,
skipping route restart");
+ return;
+ }
+ String routeId = getRouteId();
+ LOG.info("Stopping route {} for restart after resubscribe
failure", routeId);
+
getEndpoint().getCamelContext().getRouteController().stopRoute(routeId);
+ if (!getEndpoint().getCamelContext().isRunAllowed()) {
+ LOG.debug("Camel context is stopping, not restarting
route {}", routeId);
+ return;
+ }
+ 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(restartExecutor);
+ }
+ });
+ } catch (RuntimeException e) {
+ restarting.set(false);
+ if (executor != null) {
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ }
+ getExceptionHandler().handleException(
+ "Failed to schedule route restart after resubscribe
failure", e);
+ }
+ }
+
@Override
protected void doStop() throws Exception {
MqttClient ownedClient = stopClient ? client : null;
diff --git
a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoResubscribeFailureTest.java
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoResubscribeFailureTest.java
new file mode 100644
index 000000000000..612304586bc8
--- /dev/null
+++
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoResubscribeFailureTest.java
@@ -0,0 +1,250 @@
+/*
+ * 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;
+
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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.client.mqttv3.MqttCallbackExtended;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.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.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PahoResubscribeFailureTest 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() {
+ }
+ };
+ }
+
+ @Test
+ void successfulResubscribeOnReconnectShouldKeepRouteStarted() throws
Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttCallbackExtended callback = startRouteWithExternalClient(client);
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+ verify(client, times(2)).subscribe("test", 2);
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+ }
+
+ @Test
+ void
resubscribeFailureWithOwnedClientShouldRestartRouteAndReplaceOwnedClient()
throws Exception {
+ MqttClient failedClient = connectedClient();
+ MqttClient recoveredClient = connectedClient();
+ AtomicInteger createdClients = new AtomicInteger();
+ MqttCallbackExtended callback =
startRouteWithOwnedClients(createdClients, failedClient, recoveredClient);
+ doThrow(new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION))
+ .when(failedClient).subscribe(anyString(), anyInt());
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+ assertThat(createdClients).hasValue(2);
+ verify(failedClient).close(true);
+ verify(recoveredClient).connect(any(MqttConnectOptions.class));
+ verify(recoveredClient).subscribe("test", 2);
+ verify(recoveredClient, never()).close(true);
+ });
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+ assertThat(createdClients).hasValue(2);
+ verify(failedClient, times(2)).subscribe("test", 2);
+ }
+
+ @Test
+ void resubscribeFailureWithExternalClientShouldNotRestartOrCloseClient()
throws Exception {
+ MqttClient client = connectedClient();
+ MqttCallbackExtended callback = startRouteWithExternalClient(client);
+ doThrow(new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION))
+ .when(client).subscribe(anyString(), anyInt());
+
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+ verify(client, times(2)).subscribe("test", 2);
+ verify(client, never()).connect(any(MqttConnectOptions.class));
+ verify(client, never()).close(true);
+ }
+
+ @Test
+ void concurrentReconnectFailuresShouldTriggerOnlyOneRestart() throws
Exception {
+ MqttClient failedClient = connectedClient();
+ MqttClient recoveredClient = connectedClient();
+ AtomicInteger createdClients = new AtomicInteger();
+ MqttCallbackExtended callback =
startRouteWithOwnedClients(createdClients, failedClient, recoveredClient);
+ CyclicBarrier subscribeBarrier = new CyclicBarrier(2);
+ CountDownLatch closeEntered = new CountDownLatch(1);
+ CountDownLatch releaseClose = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ subscribeBarrier.await(5, TimeUnit.SECONDS);
+ throw new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+ }).when(failedClient).subscribe(anyString(), anyInt());
+ doAnswer(invocation -> {
+ closeEntered.countDown();
+ assertThat(releaseClose.await(5, TimeUnit.SECONDS)).isTrue();
+ return null;
+ }).when(failedClient).close(true);
+
+ CompletableFuture<Void> first = CompletableFuture.runAsync(
+ () -> callback.connectComplete(true, "tcp://localhost:1883"));
+ CompletableFuture<Void> second = CompletableFuture.runAsync(
+ () -> callback.connectComplete(true, "tcp://localhost:1883"));
+
+ assertThat(closeEntered.await(5, TimeUnit.SECONDS)).isTrue();
+ first.get(5, TimeUnit.SECONDS);
+ second.get(5, TimeUnit.SECONDS);
+ releaseClose.countDown();
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+ assertThat(createdClients).hasValue(2);
+ verify(failedClient).close(true);
+ verify(recoveredClient).connect(any(MqttConnectOptions.class));
+ });
+ }
+
+ @Test
+ void reconnectCallbackAfterShutdownShouldNotResubscribeOrRestart() throws
Exception {
+ MqttClient client = connectedClient();
+ AtomicInteger createdClients = new AtomicInteger();
+ MqttCallbackExtended callback =
startRouteWithOwnedClients(createdClients, client);
+
+ context.getRouteController().stopRoute(ROUTE_ID);
+ callback.connectComplete(true, "tcp://localhost:1883");
+
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Stopped);
+ assertThat(createdClients).hasValue(1);
+ verify(client, times(1)).subscribe("test", 2);
+ verify(client).close(true);
+ }
+
+ @Test
+ void initialConnectShouldNotResubscribe() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttCallbackExtended callback = startRouteWithExternalClient(client);
+
+ callback.connectComplete(false, "tcp://localhost:1883");
+
+ verify(client, times(1)).subscribe("test", 2);
+
assertThat(context.getRouteController().getRouteStatus(ROUTE_ID)).isEqualTo(ServiceStatus.Started);
+ }
+
+ private MqttCallbackExtended startRouteWithExternalClient(MqttClient
client) throws Exception {
+ PahoEndpoint endpoint = context.getEndpoint(
+ "paho:test?brokerUrl=tcp://localhost:1883",
PahoEndpoint.class);
+ endpoint.setClient(client);
+
context.addRoutes(createRoute("paho:test?brokerUrl=tcp://localhost:1883"));
+ context.start();
+ return captureCallback(client);
+ }
+
+ private MqttCallbackExtended startRouteWithOwnedClients(
+ AtomicInteger createdClients, MqttClient... clients)
+ throws Exception {
+ PahoConfiguration configuration = new PahoConfiguration();
+ configuration.setAutomaticReconnect(true);
+ configuration.setBrokerUrl("tcp://localhost:1883");
+ Deque<MqttClient> availableClients = new
ArrayDeque<>(Arrays.asList(clients));
+ PahoComponent component = new PahoComponent(context) {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ PahoEndpoint endpoint = new PahoEndpoint(uri, remaining, this,
configuration.copy()) {
+ @Override
+ public Consumer createConsumer(Processor processor) throws
Exception {
+ PahoConsumer consumer = new PahoConsumer(this,
processor) {
+ @Override
+ MqttClient createClient() {
+ createdClients.incrementAndGet();
+ return availableClients.removeFirst();
+ }
+ };
+ configureConsumer(consumer);
+ return consumer;
+ }
+ };
+ return endpoint;
+ }
+ };
+ context.addComponent("paho-owned", component);
+ context.addRoutes(createRoute("paho-owned:test"));
+ context.start();
+ return captureCallback(clients[0]);
+ }
+
+ private RouteBuilder createRoute(String uri) {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(uri).id(ROUTE_ID).to("mock:result");
+ }
+ };
+ }
+
+ private MqttCallbackExtended captureCallback(MqttClient client) throws
Exception {
+ ArgumentCaptor<MqttCallbackExtended> callbackCaptor =
ArgumentCaptor.forClass(MqttCallbackExtended.class);
+ verify(client).setCallback(callbackCaptor.capture());
+ return callbackCaptor.getValue();
+ }
+
+ private static MqttClient connectedClient() {
+ MqttClient client = mock(MqttClient.class);
+ when(client.isConnected()).thenReturn(true);
+ return client;
+ }
+}
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 a1256c0d26db..79d921a41d34 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
@@ -1014,6 +1014,28 @@ Deployments that relied on the previous behaviour — a
development cluster with
certificate, for example — must either configure a truststore or set
`camel.knative.client.ssl.trust.all`
explicitly. `KnativeOidcClientOptions` extends this class and is affected the
same way.
+=== camel-paho
+
+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. With
`cleanSession=true`, the consumer
+unsubscribes before disconnecting; with `cleanSession=false`, it keeps the
durable subscription. 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
+----
+
=== camel-paho-mqtt5
When `automaticReconnect=true` and the MQTT broker reconnects, the consumer
now restarts the route
