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 2a2a2f747069 CAMEL-24465: Fix Paho client cleanup after failed startup
2a2a2f747069 is described below
commit 2a2a2f7470696b91edd9b0486b19203f5fcca682
Author: nkokitkar <[email protected]>
AuthorDate: Tue Aug 25 05:19:57 2026 -0700
CAMEL-24465: Fix Paho client cleanup after failed startup
Force-close internally owned MQTT v3 and v5 clients when a consumer's
startup or shutdown fails, including clients already disconnected or
connected before a later startup step fails. This stops repeated
recovery attempts from leaking Paho client threads and persistence
contexts. Externally supplied/shared clients remain caller-owned and
are never closed. When lifecycle cleanup also fails, the original
startup/shutdown failure stays primary and the close failure is
attached as suppressed. Scope is limited to consumers, matching the
source fix. Focused lifecycle tests cover failed startup, disconnected
and durable clients, shared clients, and cleanup failures for both
Paho versions.
Closes #25508
Co-authored-by: Copilot App <[email protected]>
---
components/camel-paho-mqtt5/pom.xml | 5 +
.../component/paho/mqtt5/PahoMqtt5Consumer.java | 181 +++++++++++++--------
.../paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java | 158 ++++++++++++++++++
components/camel-paho/pom.xml | 5 +
.../apache/camel/component/paho/PahoConsumer.java | 165 ++++++++++++-------
.../component/paho/PahoConsumerLifecycleTest.java | 158 ++++++++++++++++++
6 files changed, 546 insertions(+), 126 deletions(-)
diff --git a/components/camel-paho-mqtt5/pom.xml
b/components/camel-paho-mqtt5/pom.xml
index 95213382a046..5ce845becdf6 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..0f1b5faf661a 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
@@ -58,91 +58,138 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
protected void doStart() throws Exception {
super.doStart();
- connectionOptions = getEndpoint().createMqttConnectionOptions();
-
- if (client == null) {
- clientId = getEndpoint().getConfiguration().getClientId();
- if (clientId == null) {
- clientId = PahoMqtt5Endpoint.generateClientId();
- }
- stopClient = true;
- client = new MqttClient(
- getEndpoint().getConfiguration().getBrokerUrl(),
- clientId,
-
PahoMqtt5Endpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
- LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
- if (getEndpoint().getConfiguration().isManualAcksEnabled()) {
- client.setManualAcks(true);
-
+ stopClient = client == null;
+ try {
+ connectionOptions = getEndpoint().createMqttConnectionOptions();
+
+ if (stopClient) {
+ clientId = getEndpoint().getConfiguration().getClientId();
+ if (clientId == null) {
+ clientId = PahoMqtt5Endpoint.generateClientId();
+ }
+ client = createClient();
+ LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
+ if (getEndpoint().getConfiguration().isManualAcksEnabled()) {
+ client.setManualAcks(true);
+ }
+ client.connect(connectionOptions);
}
- client.connect(connectionOptions);
- }
- client.setCallback(new MqttCallback() {
+ client.setCallback(new MqttCallback() {
- @Override
- public void connectComplete(boolean reconnect, String serverURI) {
- if (reconnect) {
- try {
- client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
- } catch (MqttException e) {
- LOG.error("MQTT resubscribe failed {}",
e.getMessage(), e);
+ @Override
+ public void connectComplete(boolean reconnect, String
serverURI) {
+ if (reconnect) {
+ try {
+ client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ } catch (MqttException e) {
+ LOG.error("MQTT resubscribe failed {}",
e.getMessage(), e);
+ }
}
}
- }
- @Override
- public void authPacketArrived(int reasonCode, MqttProperties
properties) {
- LOG.debug("Auth packet arrived {} {}", reasonCode, properties);
- }
+ @Override
+ public void authPacketArrived(int reasonCode, MqttProperties
properties) {
+ LOG.debug("Auth packet arrived {} {}", reasonCode,
properties);
+ }
- @Override
- public void disconnected(MqttDisconnectResponse response) {
- LOG.debug("MQTT broker disconnected due {}",
response.getReasonString(), response.getException());
- }
+ @Override
+ public void disconnected(MqttDisconnectResponse response) {
+ LOG.debug("MQTT broker disconnected due {}",
response.getReasonString(), response.getException());
+ }
- @Override
- public void mqttErrorOccurred(MqttException exception) {
- LOG.debug("Error occurred {}", exception.getMessage(),
exception);
- }
+ @Override
+ public void mqttErrorOccurred(MqttException exception) {
+ LOG.debug("Error occurred {}", exception.getMessage(),
exception);
+ }
+
+ @Override
+ public void messageArrived(String topic, MqttMessage message)
throws Exception {
+ LOG.debug("Message arrived on topic: {} -> {}", topic,
message);
+ Exchange exchange = createExchange(message, topic);
- @Override
- public void messageArrived(String topic, MqttMessage message)
throws Exception {
- LOG.debug("Message arrived on topic: {} -> {}", topic,
message);
- Exchange exchange = createExchange(message, topic);
+ // use default consumer callback
+ AsyncCallback cb = defaultConsumerCallback(exchange, true);
+ getAsyncProcessor().process(exchange, cb);
+ }
- // use default consumer callback
- AsyncCallback cb = defaultConsumerCallback(exchange, true);
- getAsyncProcessor().process(exchange, cb);
- }
+ @Override
+ public void deliveryComplete(IMqttToken token) {
+ LOG.debug("Delivery complete. Token: {}", token);
+ }
+ });
- @Override
- public void deliveryComplete(IMqttToken token) {
- LOG.debug("Delivery complete. Token: {}", token);
+ LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
+ client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ } catch (Exception startException) {
+ MqttClient ownedClient = stopClient ? client : null;
+ if (ownedClient != null) {
+ client = null;
+ stopClient = false;
+ if (ownedClient.isConnected()) {
+ try {
+ ownedClient.disconnect();
+ } catch (Exception disconnectException) {
+ startException.addSuppressed(disconnectException);
+ }
+ }
+ closeOwnedClient(ownedClient, startException);
}
- });
-
- LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
- client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ throw startException;
+ }
}
@Override
protected void doStop() throws Exception {
- super.doStop();
-
- if (stopClient && client != null && client.isConnected()) {
- String topic = getEndpoint().getTopic();
- // only unsubscribe if we are not durable
- if (getEndpoint().getConfiguration().isCleanStart()) {
- LOG.debug("Unsubscribing client: {} from topic: {}", clientId,
topic);
- client.unsubscribe(topic);
- } else {
- LOG.debug("Client: {} is durable so will not unsubscribe from
topic: {}", clientId, topic);
+ MqttClient ownedClient = stopClient ? client : null;
+ Exception stopException = null;
+ try {
+ super.doStop();
+
+ if (ownedClient != null && ownedClient.isConnected()) {
+ String topic = getEndpoint().getTopic();
+ // only unsubscribe if we are not durable
+ if (getEndpoint().getConfiguration().isCleanStart()) {
+ LOG.debug("Unsubscribing client: {} from topic: {}",
clientId, topic);
+ ownedClient.unsubscribe(topic);
+ } else {
+ LOG.debug("Client: {} is durable so will not unsubscribe
from topic: {}", clientId, topic);
+ }
+ LOG.debug("Disconnecting client: {} from broker: {}", clientId,
+ getEndpoint().getConfiguration().getBrokerUrl());
+ ownedClient.disconnect();
+ }
+ } catch (Exception e) {
+ stopException = e;
+ } finally {
+ client = null;
+ stopClient = false;
+ if (ownedClient != null) {
+ stopException = closeOwnedClient(ownedClient, stopException);
+ }
+ }
+ if (stopException != null) {
+ throw stopException;
+ }
+ }
+
+ MqttClient createClient() throws MqttException {
+ return new MqttClient(
+ getEndpoint().getConfiguration().getBrokerUrl(),
+ clientId,
+
PahoMqtt5Endpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
+ }
+
+ private Exception closeOwnedClient(MqttClient ownedClient, Exception
primaryException) {
+ try {
+ ownedClient.close(true);
+ } catch (Exception closeException) {
+ if (primaryException == null) {
+ return closeException;
}
- LOG.debug("Disconnecting client: {} from broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
- client.disconnect();
+ primaryException.addSuppressed(closeException);
}
- client = null;
+ return primaryException;
}
@Override
diff --git
a/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java
new file mode 100644
index 000000000000..3e5b19834076
--- /dev/null
+++
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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 org.apache.camel.CamelContext;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.Processor;
+import org.apache.camel.spi.ExchangeFactory;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowableOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PahoMqtt5ConsumerLifecycleTest {
+
+ @Test
+ void failedStartForceClosesOwnedClient() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException connectException = new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+
doThrow(connectException).when(client).connect(any(MqttConnectionOptions.class));
+ PahoMqtt5Consumer consumer = createConsumer(new
PahoMqtt5Configuration(), client);
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStart);
+
+ assertThat(thrown).isSameAs(connectException);
+ verify(client).close(true);
+ }
+
+ @Test
+ void failedStartSuppressesCloseFailure() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException connectException = new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+ MqttException closeException = new MqttException(1);
+
doThrow(connectException).when(client).connect(any(MqttConnectionOptions.class));
+ doThrow(closeException).when(client).close(true);
+ PahoMqtt5Consumer consumer = createConsumer(new
PahoMqtt5Configuration(), client);
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStart);
+
+ assertThat(thrown).isSameAs(connectException);
+ assertThat(thrown.getSuppressed()).containsExactly(closeException);
+ }
+
+ @Test
+ void failedStartAfterConnectDisconnectsAndClosesOwnedClient() throws
Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException subscribeException = new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+ PahoMqtt5Configuration configuration = new PahoMqtt5Configuration();
+ when(client.isConnected()).thenReturn(true);
+ doThrow(subscribeException).when(client).subscribe("test",
configuration.getQos());
+ PahoMqtt5Consumer consumer = createConsumer(configuration, client);
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStart);
+
+ assertThat(thrown).isSameAs(subscribeException);
+ verify(client).disconnect();
+ verify(client).close(true);
+ }
+
+ @Test
+ void stopForceClosesOwnedClientWhenDisconnected() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ PahoMqtt5Consumer consumer = createConsumer(new
PahoMqtt5Configuration(), client);
+ consumer.doStart();
+
+ consumer.doStop();
+
+ verify(client, never()).disconnect();
+ verify(client).close(true);
+ }
+
+ @Test
+ void failedStopSuppressesCloseFailure() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException disconnectException = new MqttException(1);
+ MqttException closeException = new MqttException(2);
+ when(client.isConnected()).thenReturn(true);
+ doThrow(disconnectException).when(client).disconnect();
+ doThrow(closeException).when(client).close(true);
+ PahoMqtt5Configuration configuration = new PahoMqtt5Configuration();
+ configuration.setCleanStart(false);
+ PahoMqtt5Consumer consumer = createConsumer(configuration, client);
+ consumer.doStart();
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStop);
+
+ assertThat(thrown).isSameAs(disconnectException);
+ assertThat(thrown.getSuppressed()).containsExactly(closeException);
+ }
+
+ @Test
+ void durableConnectedClientDisconnectsAndClosesWithoutUnsubscribe() throws
Exception {
+ MqttClient client = mock(MqttClient.class);
+ when(client.isConnected()).thenReturn(true);
+ PahoMqtt5Configuration configuration = new PahoMqtt5Configuration();
+ configuration.setCleanStart(false);
+ PahoMqtt5Consumer consumer = createConsumer(configuration, client);
+ consumer.doStart();
+
+ consumer.doStop();
+
+ verify(client, never()).unsubscribe("test");
+ verify(client).disconnect();
+ verify(client).close(true);
+ }
+
+ @Test
+ void sharedClientIsNotClosed() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ PahoMqtt5Consumer consumer = createConsumer(new
PahoMqtt5Configuration(), mock(MqttClient.class));
+ consumer.setClient(client);
+
+ consumer.doStart();
+ consumer.doStop();
+
+ verify(client, never()).close(true);
+ }
+
+ private static PahoMqtt5Consumer createConsumer(PahoMqtt5Configuration
configuration, MqttClient createdClient) {
+ CamelContext context = mock(CamelContext.class);
+ ExtendedCamelContext extension = mock(ExtendedCamelContext.class);
+ ExchangeFactory exchangeFactory = mock(ExchangeFactory.class);
+ when(context.getCamelContextExtension()).thenReturn(extension);
+ when(extension.getExchangeFactory()).thenReturn(exchangeFactory);
+
when(exchangeFactory.newExchangeFactory(any())).thenReturn(exchangeFactory);
+ PahoMqtt5Endpoint endpoint = new PahoMqtt5Endpoint(
+ "paho-mqtt5:test", "test", new PahoMqtt5Component(context),
configuration);
+ return new PahoMqtt5Consumer(endpoint, mock(Processor.class)) {
+ @Override
+ MqttClient createClient() {
+ return createdClient;
+ }
+ };
+ }
+}
diff --git a/components/camel-paho/pom.xml b/components/camel-paho/pom.xml
index 3c4c7895b717..d5804f694efa 100644
--- a/components/camel-paho/pom.xml
+++ b/components/camel-paho/pom.xml
@@ -54,6 +54,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>
<groupId>org.apache.camel</groupId>
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 4bde73ec28e8..19269c2cc05e 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
@@ -57,81 +57,128 @@ public class PahoConsumer extends DefaultConsumer {
protected void doStart() throws Exception {
super.doStart();
- connectOptions =
PahoEndpoint.createMqttConnectOptions(getEndpoint().getConfiguration());
-
- if (client == null) {
- clientId = getEndpoint().getConfiguration().getClientId();
- if (clientId == null) {
- clientId = "camel-" + MqttClient.generateClientId();
+ stopClient = client == null;
+ try {
+ connectOptions =
PahoEndpoint.createMqttConnectOptions(getEndpoint().getConfiguration());
+
+ if (stopClient) {
+ clientId = getEndpoint().getConfiguration().getClientId();
+ if (clientId == null) {
+ clientId = "camel-" + MqttClient.generateClientId();
+ }
+ client = createClient();
+ LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
+ if (getEndpoint().getConfiguration().isManualAcksEnabled()) {
+ client.setManualAcks(true);
+ }
+ client.connect(connectOptions);
}
- stopClient = true;
- client = new MqttClient(
- getEndpoint().getConfiguration().getBrokerUrl(),
- clientId,
-
PahoEndpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
- LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
- if (getEndpoint().getConfiguration().isManualAcksEnabled()) {
- client.setManualAcks(true);
- }
- client.connect(connectOptions);
- }
+ client.setCallback(new MqttCallbackExtended() {
- client.setCallback(new MqttCallbackExtended() {
-
- @Override
- public void connectComplete(boolean reconnect, String serverURI) {
- if (reconnect) {
- try {
- client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
- } catch (MqttException e) {
- LOG.error("MQTT resubscribe failed {}",
e.getMessage(), e);
+ @Override
+ public void connectComplete(boolean reconnect, String
serverURI) {
+ if (reconnect) {
+ try {
+ client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ } catch (MqttException e) {
+ LOG.error("MQTT resubscribe failed {}",
e.getMessage(), e);
+ }
}
}
- }
- @Override
- public void connectionLost(Throwable cause) {
- LOG.debug("MQTT broker connection lost due {}",
cause.getMessage(), cause);
- }
+ @Override
+ public void connectionLost(Throwable cause) {
+ LOG.debug("MQTT broker connection lost due {}",
cause.getMessage(), cause);
+ }
- @Override
- public void messageArrived(String topic, MqttMessage message)
throws Exception {
- LOG.debug("Message arrived on topic: {} -> {}", topic,
message);
- Exchange exchange = createExchange(message, topic);
+ @Override
+ public void messageArrived(String topic, MqttMessage message)
throws Exception {
+ LOG.debug("Message arrived on topic: {} -> {}", topic,
message);
+ Exchange exchange = createExchange(message, topic);
- // use default consumer callback
- AsyncCallback cb = defaultConsumerCallback(exchange, true);
- getAsyncProcessor().process(exchange, cb);
- }
+ // use default consumer callback
+ AsyncCallback cb = defaultConsumerCallback(exchange, true);
+ getAsyncProcessor().process(exchange, cb);
+ }
- @Override
- public void deliveryComplete(IMqttDeliveryToken token) {
- LOG.debug("Delivery complete. Token: {}", token);
- }
- });
+ @Override
+ public void deliveryComplete(IMqttDeliveryToken token) {
+ LOG.debug("Delivery complete. Token: {}", token);
+ }
+ });
- LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
- client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
+ client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ } catch (Exception startException) {
+ MqttClient ownedClient = stopClient ? client : null;
+ if (ownedClient != null) {
+ client = null;
+ stopClient = false;
+ if (ownedClient.isConnected()) {
+ try {
+ ownedClient.disconnect();
+ } catch (Exception disconnectException) {
+ startException.addSuppressed(disconnectException);
+ }
+ }
+ closeOwnedClient(ownedClient, startException);
+ }
+ throw startException;
+ }
}
@Override
protected void doStop() throws Exception {
- super.doStop();
-
- if (stopClient && client != null && client.isConnected()) {
- String topic = getEndpoint().getTopic();
- // only unsubscribe if we are not durable
- if (getEndpoint().getConfiguration().isCleanSession()) {
- LOG.debug("Unsubscribing client: {} from topic: {}", clientId,
topic);
- client.unsubscribe(topic);
- } else {
- LOG.debug("Client: {} is durable so will not unsubscribe from
topic: {}", clientId, topic);
+ MqttClient ownedClient = stopClient ? client : null;
+ Exception stopException = null;
+ try {
+ super.doStop();
+
+ if (ownedClient != null && ownedClient.isConnected()) {
+ String topic = getEndpoint().getTopic();
+ // only unsubscribe if we are not durable
+ if (getEndpoint().getConfiguration().isCleanSession()) {
+ LOG.debug("Unsubscribing client: {} from topic: {}",
clientId, topic);
+ ownedClient.unsubscribe(topic);
+ } else {
+ LOG.debug("Client: {} is durable so will not unsubscribe
from topic: {}", clientId, topic);
+ }
+ LOG.debug("Disconnecting client: {} from broker: {}", clientId,
+ getEndpoint().getConfiguration().getBrokerUrl());
+ ownedClient.disconnect();
+ }
+ } catch (Exception e) {
+ stopException = e;
+ } finally {
+ client = null;
+ stopClient = false;
+ if (ownedClient != null) {
+ stopException = closeOwnedClient(ownedClient, stopException);
+ }
+ }
+ if (stopException != null) {
+ throw stopException;
+ }
+ }
+
+ MqttClient createClient() throws MqttException {
+ return new MqttClient(
+ getEndpoint().getConfiguration().getBrokerUrl(),
+ clientId,
+
PahoEndpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
+ }
+
+ private Exception closeOwnedClient(MqttClient ownedClient, Exception
primaryException) {
+ try {
+ ownedClient.close(true);
+ } catch (Exception closeException) {
+ if (primaryException == null) {
+ return closeException;
}
- LOG.debug("Disconnecting client: {} from broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
- client.disconnect();
+ primaryException.addSuppressed(closeException);
}
- client = null;
+ return primaryException;
}
@Override
diff --git
a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerLifecycleTest.java
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerLifecycleTest.java
new file mode 100644
index 000000000000..05ccce1025a0
--- /dev/null
+++
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerLifecycleTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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 org.apache.camel.CamelContext;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.Processor;
+import org.apache.camel.spi.ExchangeFactory;
+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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowableOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PahoConsumerLifecycleTest {
+
+ @Test
+ void failedStartForceClosesOwnedClient() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException connectException = new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+
doThrow(connectException).when(client).connect(any(MqttConnectOptions.class));
+ PahoConsumer consumer = createConsumer(new PahoConfiguration(),
client);
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStart);
+
+ assertThat(thrown).isSameAs(connectException);
+ verify(client).close(true);
+ }
+
+ @Test
+ void failedStartSuppressesCloseFailure() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException connectException = new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+ MqttException closeException = new
MqttException(MqttException.REASON_CODE_CLIENT_DISCONNECTING);
+
doThrow(connectException).when(client).connect(any(MqttConnectOptions.class));
+ doThrow(closeException).when(client).close(true);
+ PahoConsumer consumer = createConsumer(new PahoConfiguration(),
client);
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStart);
+
+ assertThat(thrown).isSameAs(connectException);
+ assertThat(thrown.getSuppressed()).containsExactly(closeException);
+ }
+
+ @Test
+ void failedStartAfterConnectDisconnectsAndClosesOwnedClient() throws
Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException subscribeException = new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+ PahoConfiguration configuration = new PahoConfiguration();
+ when(client.isConnected()).thenReturn(true);
+ doThrow(subscribeException).when(client).subscribe("test",
configuration.getQos());
+ PahoConsumer consumer = createConsumer(configuration, client);
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStart);
+
+ assertThat(thrown).isSameAs(subscribeException);
+ verify(client).disconnect();
+ verify(client).close(true);
+ }
+
+ @Test
+ void stopForceClosesOwnedClientWhenDisconnected() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ PahoConsumer consumer = createConsumer(new PahoConfiguration(),
client);
+ consumer.doStart();
+
+ consumer.doStop();
+
+ verify(client, never()).disconnect();
+ verify(client).close(true);
+ }
+
+ @Test
+ void failedStopSuppressesCloseFailure() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ MqttException disconnectException = new
MqttException(MqttException.REASON_CODE_CLIENT_DISCONNECTING);
+ MqttException closeException = new
MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+ when(client.isConnected()).thenReturn(true);
+ doThrow(disconnectException).when(client).disconnect();
+ doThrow(closeException).when(client).close(true);
+ PahoConfiguration configuration = new PahoConfiguration();
+ configuration.setCleanSession(false);
+ PahoConsumer consumer = createConsumer(configuration, client);
+ consumer.doStart();
+
+ MqttException thrown = catchThrowableOfType(MqttException.class,
consumer::doStop);
+
+ assertThat(thrown).isSameAs(disconnectException);
+ assertThat(thrown.getSuppressed()).containsExactly(closeException);
+ }
+
+ @Test
+ void durableConnectedClientDisconnectsAndClosesWithoutUnsubscribe() throws
Exception {
+ MqttClient client = mock(MqttClient.class);
+ when(client.isConnected()).thenReturn(true);
+ PahoConfiguration configuration = new PahoConfiguration();
+ configuration.setCleanSession(false);
+ PahoConsumer consumer = createConsumer(configuration, client);
+ consumer.doStart();
+
+ consumer.doStop();
+
+ verify(client, never()).unsubscribe("test");
+ verify(client).disconnect();
+ verify(client).close(true);
+ }
+
+ @Test
+ void sharedClientIsNotClosed() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ PahoConsumer consumer = createConsumer(new PahoConfiguration(),
mock(MqttClient.class));
+ consumer.setClient(client);
+
+ consumer.doStart();
+ consumer.doStop();
+
+ verify(client, never()).close(true);
+ }
+
+ private static PahoConsumer createConsumer(PahoConfiguration
configuration, MqttClient createdClient) {
+ CamelContext context = mock(CamelContext.class);
+ ExtendedCamelContext extension = mock(ExtendedCamelContext.class);
+ ExchangeFactory exchangeFactory = mock(ExchangeFactory.class);
+ when(context.getCamelContextExtension()).thenReturn(extension);
+ when(extension.getExchangeFactory()).thenReturn(exchangeFactory);
+
when(exchangeFactory.newExchangeFactory(any())).thenReturn(exchangeFactory);
+ PahoEndpoint endpoint = new PahoEndpoint(
+ "paho:test", "test", new PahoComponent(context),
configuration);
+ return new PahoConsumer(endpoint, mock(Processor.class)) {
+ @Override
+ MqttClient createClient() {
+ return createdClient;
+ }
+ };
+ }
+}
