atiaomar1978-hub commented on code in PR #25508:
URL: https://github.com/apache/camel/pull/25508#discussion_r3818341194
##########
components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java:
##########
@@ -57,81 +57,121 @@ public void setClient(MqttClient client) {
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 = 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);
-
+ 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);
}
- 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;
+ 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) {
Review Comment:
**Blocking — disconnect before close on failed startup**
`closeOwnedClient` always calls `close(true)` without checking
`isConnected()`. In the `doStart` catch block, if `connect` succeeded but
`subscribe` (or callback setup) failed, the owned client may still be connected
— Paho can throw on `close(true)`, the error gets suppressed, and the
session/resources leak persists.
`doStop` already disconnects when connected before reaching
`closeOwnedClient`; please mirror that here:
```java
if (ownedClient.isConnected()) {
ownedClient.disconnect(); // best-effort
}
ownedClient.close(true);
```
Same applies to `PahoMqtt5Consumer`. Add a test: connect succeeds, subscribe
throws, verify disconnect + close.
##########
components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java:
##########
@@ -57,81 +57,121 @@ public void setClient(MqttClient client) {
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 = 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);
-
+ 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);
}
- 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;
+ 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;
Review Comment:
**Good — shutdown leak fixed**
Capturing `ownedClient` before `super.doStop()` and always `close(true)` in
`finally` addresses the pre-existing gap where disconnect ran but `close()`
never did. Exception suppression keeps the primary stop failure visible.
##########
components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerLifecycleTest.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;
+
+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 {
Review Comment:
**Strong lifecycle test suite**
Six focused tests cover failed start, failed stop, disconnected close,
durable no-unsubscribe, shared client, and suppressed close failures. Uses
AssertJ and package-private conventions correctly.
Gap: no test for subscribe failure after successful connect — that path
exposes the disconnect-before-close issue in `closeOwnedClient`.
##########
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) {
Review Comment:
**Same fix needed here**
Mirror the `closeOwnedClient` disconnect-before-close improvement from
`PahoConsumer` — both consumers share identical lifecycle logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
