Fix Paho client cleanup after failed startup

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>

components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java

@@ -58,91 +58,131 @@ public void setClient(MqttClient client) {
     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());
+            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;
+                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().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

components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java

@@ -0,0 +1,142 @@
+/*
+ * 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(consumer::doStart, MqttException.class);
+
+        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(consumer::doStart, MqttException.class);
+
+        assertThat(thrown).isSameAs(connectException);
+        assertThat(thrown.getSuppressed()).containsExactly(closeException);
+    }
+
+    @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(consumer::doStop, MqttException.class);
+
+        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;
+            }
+        };
+    }
+}

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>