This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/28197-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit ef37f407aeb39a8971703efc174da9f86f36ac90 Author: Guillaume Nodet <[email protected]> AuthorDate: Thu Oct 22 05:34:59 2026 +0000 [backport camel-4.22.x] CAMEL-25793: camel-paho-mqtt5 - a failed message with manualAcksEnabled no longer blocks the consumer --- .../camel/catalog/components/paho-mqtt5.json | 4 +- .../camel/catalog/docs/paho-mqtt5-component.adoc | 27 ++ components/camel-paho-mqtt5/pom.xml | 12 + .../camel/component/paho/mqtt5/paho-mqtt5.json | 4 +- .../src/main/docs/paho-mqtt5-component.adoc | 27 ++ .../paho/mqtt5/PahoMqtt5Configuration.java | 7 +- .../component/paho/mqtt5/PahoMqtt5Consumer.java | 168 ++++++++++- .../paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java | 332 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 16 + .../dsl/PahoMqtt5ComponentBuilderFactory.java | 11 +- .../dsl/PahoMqtt5EndpointBuilderFactory.java | 22 +- 11 files changed, 613 insertions(+), 17 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/paho-mqtt5.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/paho-mqtt5.json index dd096fb513b6..896ee04b6205 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/paho-mqtt5.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/paho-mqtt5.json @@ -45,7 +45,7 @@ "willRetained": { "index": 18, "kind": "property", "displayName": "Will Retained", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client u [...] "willTopic": { "index": 19, "kind": "property", "displayName": "Will Topic", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses its [...] "bridgeErrorHandler": { "index": 20, "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the lik [...] - "manualAcksEnabled": { "index": 21, "kind": "property", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defaul [...] + "manualAcksEnabled": { "index": 21, "kind": "property", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defaul [...] "lazyStartProducer": { "index": 22, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fai [...] "autowiredEnabled": { "index": 23, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching [...] "client": { "index": 24, "kind": "property", "displayName": "Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.eclipse.paho.mqttv5.client.MqttClient", "deprecated": false, "autowired": false, "secret": false, "description": "To use a shared Paho client" }, @@ -87,7 +87,7 @@ "willQos": { "index": 17, "kind": "parameter", "displayName": "Will Qos", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses [...] "willRetained": { "index": 18, "kind": "parameter", "displayName": "Will Retained", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client [...] "willTopic": { "index": 19, "kind": "parameter", "displayName": "Will Topic", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses it [...] - "manualAcksEnabled": { "index": 20, "kind": "parameter", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defau [...] + "manualAcksEnabled": { "index": 20, "kind": "parameter", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defau [...] "bridgeErrorHandler": { "index": 21, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming [...] "exceptionHandler": { "index": 22, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By de [...] "exchangePattern": { "index": 23, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/paho-mqtt5-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/paho-mqtt5-component.adoc index f32ee3c3676f..a617b3f0b316 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/paho-mqtt5-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/paho-mqtt5-component.adoc @@ -79,6 +79,33 @@ String payload = "message"; producerTemplate.sendBody("paho-mqtt5:topic", payload); ---- +=== Manual acknowledgements + +With `manualAcksEnabled=true` a message is acknowledged when its exchange completes. MQTT has no negative +acknowledgement: the broker sends an unacknowledged message again only when the client connects again to a session +that the broker kept, and it does not send more than `receiveMaximum` unacknowledged messages to the client. The +client holds back only the acknowledgement of a QoS 1 message; it completes the QoS 2 flow on its own when the message +arrives, so the broker never sends a QoS 2 message again. So when an exchange fails: + +* for a QoS 1 message on a session that the broker keeps after the disconnect (`cleanStart=false` and + `sessionExpiryInterval` greater than 0) the consumer disconnects and connects again, and the broker sends the + message again; +* in all other cases (a clean session, which is the default, a QoS 2 message, or `cleanStart=false` without + `sessionExpiryInterval`, where the session ends with the connection) the broker never sends the message again, so + the consumer acknowledges it, and the failure is left to the Camel error handler; +* with a client set by the `client` option, a QoS 1 message on a session that the broker keeps stays unacknowledged + until that client connects again. The consumer does not connect that client, so it tells the cases apart with the + `cleanStart` and `sessionExpiryInterval` of the endpoint: set them to match the connection options of the client. + With the defaults (a clean session) a failed message is acknowledged. + +The reconnects of failed exchanges in a row back off: the first one waits one second, each further one waits twice as +long, up to `maxReconnectDelay` (but at least one second), and any exchange that succeeds, also one of another +message, starts again from one second. So on a topic where other messages succeed in between, a message that always +fails makes the consumer reconnect about once a second. +Each such reconnect is logged at WARN. The broker then also sends again the other QoS 1 messages that were not +acknowledged yet, such as the ones still being routed. A message that always fails is still delivered again and again; +use a Camel error handler, for example a dead letter channel, for such messages. + == Examples diff --git a/components/camel-paho-mqtt5/pom.xml b/components/camel-paho-mqtt5/pom.xml index 4a9ca238cb78..a6d409b87fd7 100644 --- a/components/camel-paho-mqtt5/pom.xml +++ b/components/camel-paho-mqtt5/pom.xml @@ -65,6 +65,18 @@ <version>${project.version}</version> <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.artemis</groupId> + <artifactId>artemis-mqtt-protocol</artifactId> + <version>${activemq-artemis-version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-test-infra-artemis</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> </dependencies> </project> diff --git a/components/camel-paho-mqtt5/src/generated/resources/META-INF/org/apache/camel/component/paho/mqtt5/paho-mqtt5.json b/components/camel-paho-mqtt5/src/generated/resources/META-INF/org/apache/camel/component/paho/mqtt5/paho-mqtt5.json index dd096fb513b6..896ee04b6205 100644 --- a/components/camel-paho-mqtt5/src/generated/resources/META-INF/org/apache/camel/component/paho/mqtt5/paho-mqtt5.json +++ b/components/camel-paho-mqtt5/src/generated/resources/META-INF/org/apache/camel/component/paho/mqtt5/paho-mqtt5.json @@ -45,7 +45,7 @@ "willRetained": { "index": 18, "kind": "property", "displayName": "Will Retained", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client u [...] "willTopic": { "index": 19, "kind": "property", "displayName": "Will Topic", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses its [...] "bridgeErrorHandler": { "index": 20, "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the lik [...] - "manualAcksEnabled": { "index": 21, "kind": "property", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defaul [...] + "manualAcksEnabled": { "index": 21, "kind": "property", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defaul [...] "lazyStartProducer": { "index": 22, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fai [...] "autowiredEnabled": { "index": 23, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching [...] "client": { "index": 24, "kind": "property", "displayName": "Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.eclipse.paho.mqttv5.client.MqttClient", "deprecated": false, "autowired": false, "secret": false, "description": "To use a shared Paho client" }, @@ -87,7 +87,7 @@ "willQos": { "index": 17, "kind": "parameter", "displayName": "Will Qos", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses [...] "willRetained": { "index": 18, "kind": "parameter", "displayName": "Will Retained", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client [...] "willTopic": { "index": 19, "kind": "parameter", "displayName": "Will Topic", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses it [...] - "manualAcksEnabled": { "index": 20, "kind": "parameter", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defau [...] + "manualAcksEnabled": { "index": 20, "kind": "parameter", "displayName": "Manual Acks Enabled", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.paho.mqtt5.PahoMqtt5Configuration", "configurationField": "configuration", "description": "Sets whether to use manual acknowledgements for the client. By defau [...] "bridgeErrorHandler": { "index": 21, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming [...] "exceptionHandler": { "index": 22, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By de [...] "exchangePattern": { "index": 23, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, diff --git a/components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc b/components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc index f32ee3c3676f..a617b3f0b316 100644 --- a/components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc +++ b/components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc @@ -79,6 +79,33 @@ String payload = "message"; producerTemplate.sendBody("paho-mqtt5:topic", payload); ---- +=== Manual acknowledgements + +With `manualAcksEnabled=true` a message is acknowledged when its exchange completes. MQTT has no negative +acknowledgement: the broker sends an unacknowledged message again only when the client connects again to a session +that the broker kept, and it does not send more than `receiveMaximum` unacknowledged messages to the client. The +client holds back only the acknowledgement of a QoS 1 message; it completes the QoS 2 flow on its own when the message +arrives, so the broker never sends a QoS 2 message again. So when an exchange fails: + +* for a QoS 1 message on a session that the broker keeps after the disconnect (`cleanStart=false` and + `sessionExpiryInterval` greater than 0) the consumer disconnects and connects again, and the broker sends the + message again; +* in all other cases (a clean session, which is the default, a QoS 2 message, or `cleanStart=false` without + `sessionExpiryInterval`, where the session ends with the connection) the broker never sends the message again, so + the consumer acknowledges it, and the failure is left to the Camel error handler; +* with a client set by the `client` option, a QoS 1 message on a session that the broker keeps stays unacknowledged + until that client connects again. The consumer does not connect that client, so it tells the cases apart with the + `cleanStart` and `sessionExpiryInterval` of the endpoint: set them to match the connection options of the client. + With the defaults (a clean session) a failed message is acknowledged. + +The reconnects of failed exchanges in a row back off: the first one waits one second, each further one waits twice as +long, up to `maxReconnectDelay` (but at least one second), and any exchange that succeeds, also one of another +message, starts again from one second. So on a topic where other messages succeed in between, a message that always +fails makes the consumer reconnect about once a second. +Each such reconnect is logged at WARN. The broker then also sends again the other QoS 1 messages that were not +acknowledged yet, such as the ones still being routed. A message that always fails is still delivered again and again; +use a Camel error handler, for example a dead letter channel, for such messages. + == Examples diff --git a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Configuration.java b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Configuration.java index 6ef714816a0e..32da4687322b 100644 --- a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Configuration.java +++ b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Configuration.java @@ -555,8 +555,11 @@ public class PahoMqtt5Configuration implements Cloneable { * * By default, this is false and message will be automatically acknowledged upon received by Camel. If set to true, * the acknowledgement is deferred to be acknowledged by Camel at the end of processing the message. This ensures - * that only successfully processed messages is acknowledged, and allows to rollback and retry the message in case - * of an error during routing in Camel. + * that only successfully processed messages are acknowledged. When the processing of a QoS 1 message fails on a + * session that the broker keeps (cleanStart=false and sessionExpiryInterval greater than 0), the consumer connects + * again so that the broker sends the message again. In all other cases the broker would never send it again (the + * client completes the QoS 2 flow on its own, and without sessionExpiryInterval the session ends with the + * connection), so the message is acknowledged. */ public boolean isManualAcksEnabled() { return this.manualAcksEnabled; 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 deec07b82392..bdaa8b401b59 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 @@ -17,7 +17,10 @@ package org.apache.camel.component.paho.mqtt5; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.camel.AsyncCallback; import org.apache.camel.Endpoint; @@ -45,6 +48,14 @@ public class PahoMqtt5Consumer extends DefaultConsumer { private volatile boolean stopClient; private volatile MqttConnectionOptions connectionOptions; private final AtomicBoolean restarting = new AtomicBoolean(false); + private final AtomicBoolean reconnecting = new AtomicBoolean(false); + // set while the consumer connects the client again after a failed exchange: Paho reports that connect to + // connectComplete with reconnect=true once the client lost a connection on its own (it never clears its + // reconnecting flag), and the consumer subscribes again there only for the automatic reconnects of Paho + private final AtomicBoolean ownReconnect = new AtomicBoolean(false); + // failed exchanges in a row that made the consumer reconnect, for the backoff; reset by a successful exchange + private final AtomicInteger failedMessageReconnects = new AtomicInteger(); + private volatile ScheduledExecutorService reconnectExecutor; public PahoMqtt5Consumer(Endpoint endpoint, Processor processor) { super(endpoint, processor); @@ -64,7 +75,15 @@ public class PahoMqtt5Consumer extends DefaultConsumer { connectionOptions = getEndpoint().createMqttConnectionOptions(); + failedMessageReconnects.set(0); if (client == null) { + if (getEndpoint().getConfiguration().isManualAcksEnabled() && isSessionKeptAfterDisconnect()) { + // before the connect: the broker sends the unacknowledged messages of the session right after CONNACK, + // and one of them can fail and need a reconnect + reconnectExecutor = getEndpoint().getCamelContext().getExecutorServiceManager() + .newSingleThreadScheduledExecutor(this, "PahoMqtt5-Reconnect"); + } + clientId = getEndpoint().getConfiguration().getClientId(); if (clientId == null) { clientId = PahoMqtt5Endpoint.generateClientId(); @@ -83,7 +102,9 @@ public class PahoMqtt5Consumer extends DefaultConsumer { @Override public void connectComplete(boolean reconnect, String serverURI) { - if (reconnect) { + // the connect of reconnect() subscribes again itself, only when the broker did not keep the session + boolean own = ownReconnect.compareAndSet(true, false); + if (reconnect && !own) { try { client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); } catch (MqttException e) { @@ -163,6 +184,9 @@ public class PahoMqtt5Consumer extends DefaultConsumer { @Override protected void doStop() throws Exception { + shutdownReconnectExecutor(); + reconnecting.set(false); + ownReconnect.set(false); super.doStop(); if (stopClient && client != null && client.isConnected()) { @@ -180,6 +204,24 @@ public class PahoMqtt5Consumer extends DefaultConsumer { client = null; } + private void shutdownReconnectExecutor() { + ScheduledExecutorService executor = reconnectExecutor; + if (executor != null) { + reconnectExecutor = null; + getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor); + } + } + + /** + * Whether the broker keeps the session after the connection closes: cleanStart=false and a session expiry interval + * greater than 0. An absent session expiry interval (the default, -1) means 0 (MQTT 5, 3.1.2.11.2), so the session + * ends with the connection. + */ + private boolean isSessionKeptAfterDisconnect() { + PahoMqtt5Configuration configuration = getEndpoint().getConfiguration(); + return !configuration.isCleanStart() && configuration.getSessionExpiryInterval() > 0; + } + MqttClient createClient() throws MqttException { return new MqttClient( getEndpoint().getConfiguration().getBrokerUrl(), @@ -187,6 +229,127 @@ public class PahoMqtt5Consumer extends DefaultConsumer { PahoMqtt5Endpoint.createMqttClientPersistence(getEndpoint().getConfiguration())); } + /** + * MQTT has no negative acknowledgement: the broker sends an unacknowledged message again only when the client + * connects again to a session that the broker kept, and it sends no more than receiveMaximum unacknowledged + * messages to the client. Paho holds back only the PUBACK of a QoS 1 message until it is acknowledged manually; it + * completes the QoS 2 flow (PUBREC, PUBCOMP) on its own, so the broker never sends a QoS 2 message again. A failed + * QoS 1 message that is just left unacknowledged is therefore never retried, and enough of them stop the consumer + * for good. + * <p> + * For a QoS 1 message on a session that the broker keeps after the disconnect (cleanStart=false and + * sessionExpiryInterval greater than 0) the client reconnects, so that the broker sends the message again; the + * reconnects of failed exchanges in a row back off (see {@link #reconnectDelay(int)}), so that a message that + * always fails does not reconnect in a tight loop. In every other case the broker never sends the message again, so + * it is acknowledged, which frees its place in the receive window. + */ + private void rollback(Exchange exchange, MqttMessage mqttMessage) { + MqttClient current = client; + if (current == null) { + return; + } + if (mqttMessage.getQos() != 1 || !isSessionKeptAfterDisconnect()) { + LOG.debug("Acknowledging failed MQTT message with ID: {} (QoS {}) of Exchange ID: {} as the broker does not" + + " deliver it again", + mqttMessage.getId(), mqttMessage.getQos(), exchange.getExchangeId()); + try { + current.messageArrivedComplete(mqttMessage.getId(), mqttMessage.getQos()); + } catch (MqttException e) { + getExceptionHandler().handleException( + "Error acknowledging failed MQTT message with ID: " + mqttMessage.getId(), exchange, e); + } + } else if (stopClient) { + reconnectAfterFailure(current, exchange, mqttMessage); + } else { + LOG.warn("Rollback due to error processing Exchange ID: {}: MQTT message with ID: {} stays unacknowledged" + + " until the externally provided client connects again", + exchange.getExchangeId(), mqttMessage.getId()); + } + } + + private void reconnectAfterFailure(MqttClient current, Exchange exchange, MqttMessage mqttMessage) { + String reason = "Rollback due to error processing Exchange ID: " + exchange.getExchangeId() + + ", so that the broker delivers MQTT message with ID: " + mqttMessage.getId() + " again"; + if (!scheduleReconnect(current, reason)) { + LOG.debug("{}: a reconnect is already scheduled", reason); + } + } + + /** + * The delay before the reconnect of the given failed exchange in a row: the minimum automatic reconnect delay of + * the client (1 second), doubled for each further failed exchange in a row, up to maxReconnectDelay, but never less + * than the minimum (a maxReconnectDelay below it would reconnect in a tight loop). + */ + private long reconnectDelay(int attempt) { + long max = Math.max(0, getEndpoint().getConfiguration().getMaxReconnectDelay()); + MqttConnectionOptions options = connectionOptions; + long min = options != null ? TimeUnit.SECONDS.toMillis(options.getAutomaticReconnectMinDelay()) : 1000; + long delay = min << Math.min(attempt - 1, 20); + return Math.max(min, Math.min(delay, max)); + } + + private void reconnect(MqttClient current) { + try { + // the message callback may still be running (disconnect is not allowed from it), so this runs here + if (isRunAllowed() && current == client && current.isConnected()) { + current.disconnect(); + } + } catch (Exception e) { + getExceptionHandler().handleException("Failed to disconnect after a failed exchange", e); + } + // the disconnect covers every message that failed until now; a message the broker sends again on the new + // connection can fail before this method returns, and must be able to schedule the next reconnect + reconnecting.set(false); + try { + if (isRunAllowed() && current == client && !current.isConnected()) { + ownReconnect.set(true); + IMqttToken token = current.connectWithResult(connectionOptions); + // the subscription is part of the session; subscribing again would make the broker send the retained + // message of the topic again (Retain Handling 0) + if (!token.getSessionPresent()) { + LOG.warn("The broker did not keep the session of client: {}: the failed MQTT message and the other" + + " unacknowledged ones are not delivered again; subscribing again to topic: {}", + clientId, getEndpoint().getTopic()); + current.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } + } + } catch (Exception e) { + // a failed connect does not call connectComplete + ownReconnect.set(false); + getExceptionHandler().handleException("Failed to reconnect after a failed exchange", e); + // the client does not reconnect on its own after a disconnect it was asked for, so try again later + scheduleReconnect(current, "Reconnect failed"); + } + } + + /** + * Schedules a reconnect after the backoff delay, unless one is already scheduled or the consumer is stopping. + */ + private boolean scheduleReconnect(MqttClient current, String reason) { + ScheduledExecutorService executor = reconnectExecutor; + if (executor == null || !isRunAllowed() || !reconnecting.compareAndSet(false, true)) { + return false; + } + int attempt = failedMessageReconnects.incrementAndGet(); + long delay = reconnectDelay(attempt); + LOG.warn("{}: reconnecting in {} ms (reconnect {} in a row)", reason, delay, attempt); + try { + executor.schedule(() -> reconnect(current), delay, TimeUnit.MILLISECONDS); + return true; + } catch (RuntimeException e) { + // the consumer is stopping + reconnecting.set(false); + return false; + } + } + + /** + * The number of failed exchanges in a row that made the consumer reconnect. Package-private, for tests. + */ + int getFailedMessageReconnects() { + return failedMessageReconnects.get(); + } + @Override public PahoMqtt5Endpoint getEndpoint() { return (PahoMqtt5Endpoint) super.getEndpoint(); @@ -206,6 +369,7 @@ public class PahoMqtt5Consumer extends DefaultConsumer { exchange.getExchangeExtension().addOnCompletion(new SynchronizationAdapter() { @Override public void onComplete(Exchange exchange) { + failedMessageReconnects.set(0); try { PahoMqtt5Consumer.this.client.messageArrivedComplete(mqttMessage.getId(), mqttMessage.getQos()); } catch (MqttException e) { @@ -217,7 +381,7 @@ public class PahoMqtt5Consumer extends DefaultConsumer { @Override public void onFailure(Exchange exchange) { - LOG.debug("Rollback due to error processing Exchange ID: {}", exchange.getExchangeId()); + rollback(exchange, mqttMessage); } }); } diff --git a/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java new file mode 100644 index 000000000000..64f18ae24ce9 --- /dev/null +++ b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java @@ -0,0 +1,332 @@ +/* + * 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.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.test.infra.artemis.services.ArtemisService; +import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory; +import org.eclipse.paho.mqttv5.client.MqttClient; +import org.eclipse.paho.mqttv5.client.MqttConnectionOptionsBuilder; +import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * With manualAcksEnabled=true a message whose exchange failed is not acknowledged. The broker sends at most + * receiveMaximum unacknowledged QoS 1/2 messages to a client, and sends an unacknowledged message again only when the + * client connects again to the same session. These tests use a real broker (embedded Artemis, MQTT 5). + */ +class PahoMqtt5ManualAcksFailureTest { + + @RegisterExtension + static ArtemisService service = ArtemisServiceFactory.createSingletonMQTTService(); + + private final List<String> routed = new CopyOnWriteArrayList<>(); + private final Map<String, AtomicInteger> attempts = new ConcurrentHashMap<>(); + private final Map<String, List<Long>> attemptTimes = new ConcurrentHashMap<>(); + + @Test + void failedMessagesDoNotStopTheConsumer() throws Exception { + // clean session (the default) and a receive maximum of 2: after two failed exchanges the broker has two + // unacknowledged messages in flight for this client + String id = UUID.randomUUID().toString(); + String topic = "manualack/clean/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&receiveMaximum=2&clientId=manual-ack-clean-" + id; + CamelContext context = createContext(uri, 1); + try { + publish(topic, "fail-1", "fail-2", "ok-3", "ok-4"); + + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("ok-3", "ok-4")); + } finally { + context.stop(); + } + } + + @Test + void acknowledgedMessagesKeepTheReceiveWindowOpen() throws Exception { + String id = UUID.randomUUID().toString(); + String topic = "manualack/ok/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&receiveMaximum=2&clientId=manual-ack-ok-" + id; + CamelContext context = createContext(uri, 1); + try { + publish(topic, "ok-1", "ok-2", "ok-3", "ok-4"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("ok-1", "ok-2", "ok-3", "ok-4")); + } finally { + context.stop(); + } + } + + @Test + void failedMessageIsDeliveredAgainWithPersistentSession() throws Exception { + // a session that the broker keeps: the documented rollback and retry of a failed message + String id = UUID.randomUUID().toString(); + String topic = "manualack/persistent/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60" + + "&clientId=manual-ack-persistent-" + id; + CamelContext context = createContext(uri, 1); + try { + publish(topic, "fail-1", "ok-2"); + + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactlyInAnyOrder("fail-1", "ok-2")); + assertThat(attempts.get("fail-1")).hasValue(2); + } finally { + context.stop(); + } + } + + @Test + void failedQos2MessageIsAcknowledgedWithoutReconnect() throws Exception { + // Paho completes the QoS 2 flow (PUBREC, PUBCOMP) without waiting for the manual acknowledgement, so the broker + // never sends a QoS 2 message again, even to a session that it keeps: a reconnect would not retry it + String id = UUID.randomUUID().toString(); + String topic = "manualack/qos2/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=2&manualAcksEnabled=true&receiveMaximum=2&cleanStart=false&sessionExpiryInterval=60" + + "&clientId=manual-ack-qos2-" + id; + CamelContext context = createContext(uri, 1); + try { + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + publish(topic, 2, false, "fail-1", "fail-2"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { + assertThat(attempts.get("fail-1")).hasValue(1); + assertThat(attempts.get("fail-2")).hasValue(1); + }); + await().during(1500, TimeUnit.MILLISECONDS).atMost(3, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(consumer.getFailedMessageReconnects()).isZero()); + + publish(topic, 2, false, "ok-3", "ok-4"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("ok-3", "ok-4")); + assertThat(attempts.get("fail-1")).hasValue(1); + assertThat(attempts.get("fail-2")).hasValue(1); + } finally { + context.stop(); + } + } + + @Test + void failedMessageIsAcknowledgedWhenTheSessionEndsWithTheConnection() throws Exception { + // cleanStart=false without sessionExpiryInterval: the interval is not sent, which means 0, so the broker ends + // the session when the client disconnects, and a reconnect would lose the messages instead of sending them again + String id = UUID.randomUUID().toString(); + String topic = "manualack/noexpiry/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&receiveMaximum=1&cleanStart=false" + + "&clientId=manual-ack-noexpiry-" + id; + CamelContext context = createContext(uri, 1); + try { + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + publish(topic, "fail-1"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(attempts.get("fail-1")).hasValue(1)); + await().during(1500, TimeUnit.MILLISECONDS).atMost(3, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(consumer.getFailedMessageReconnects()).isZero()); + + // with a receive maximum of 1 the broker sends the next message only after fail-1 is acknowledged + publish(topic, "ok-2"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("ok-2")); + assertThat(attempts.get("fail-1")).hasValue(1); + } finally { + context.stop(); + } + } + + @Test + void reconnectToKeptSessionDoesNotSubscribeAgain() throws Exception { + // the subscription is part of the session that the broker kept; subscribing again would make the broker send + // the retained message of the topic again + String id = UUID.randomUUID().toString(); + String topic = "manualack/retained/" + id; + publish(topic, 1, true, "retained-0"); + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60" + + "&clientId=manual-ack-retained-" + id; + CamelContext context = createContext(uri, 1); + try { + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("retained-0")); + + publish(topic, "fail-1"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("retained-0", "fail-1")); + assertThat(attempts.get("fail-1")).hasValue(2); + await().during(1, TimeUnit.SECONDS).atMost(2, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(attempts.get("retained-0")).hasValue(1)); + } finally { + context.stop(); + publish(topic, 1, true); + } + } + + @Test + void reconnectAfterAnEarlierConnectionLossDoesNotSubscribeAgain() throws Exception { + // once the client lost a connection on its own, Paho reports every later connect to connectComplete as a + // reconnect: the reconnect of a failed exchange must still not subscribe again to the session the broker kept + String id = UUID.randomUUID().toString(); + String topic = "manualack/lost/" + id; + String clientId = "manual-ack-lost-" + id; + publish(topic, 1, true, "retained-0"); + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60&clientId=" + clientId; + CamelContext context = createContext(uri, 1); + try { + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("retained-0")); + + // another connection with the same client id takes the session over: the broker disconnects the consumer's + // client, which reconnects on its own (automaticReconnect) once the session is free again + MqttClient other = new MqttClient(brokerUrl(), clientId, new MemoryPersistence()); + try { + other.connect(new MqttConnectionOptionsBuilder().cleanStart(false).sessionExpiryInterval(60L).build()); + other.disconnect(); + } finally { + other.close(); + } + await().atMost(10, TimeUnit.SECONDS).until(() -> consumer.getClient().isConnected()); + // the automatic reconnect of Paho subscribes again (Retain Handling 0), so the broker sends the retained + // message again + await().atMost(10, TimeUnit.SECONDS).until(() -> attempts.get("retained-0").get() == 2); + int retainedBefore = attempts.get("retained-0").get(); + + publish(topic, "fail-1"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(attempts.get("fail-1")).hasValue(2)); + await().during(1, TimeUnit.SECONDS).atMost(2, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(attempts.get("retained-0")).hasValue(retainedBefore)); + } finally { + context.stop(); + publish(topic, 1, true); + } + } + + @Test + void reconnectsAfterFailuresInARowBackOff() throws Exception { + // a message that fails three times in a row: the reconnects wait 1 s, then 2 s capped to maxReconnectDelay + String id = UUID.randomUUID().toString(); + String topic = "manualack/backoff/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60&maxReconnectDelay=1500" + + "&clientId=manual-ack-backoff-" + id; + CamelContext context = createContext(uri, 3); + try { + publish(topic, "fail-3"); + + await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("fail-3")); + List<Long> times = attemptTimes.get("fail-3"); + assertThat(times).hasSize(4); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(1) - times.get(0))).isGreaterThanOrEqualTo(900); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(2) - times.get(1))).isGreaterThanOrEqualTo(1400); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(3) - times.get(2))).isGreaterThanOrEqualTo(1400); + // the successful exchange resets the backoff + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(consumer.getFailedMessageReconnects()).isZero()); + } finally { + context.stop(); + } + } + + @Test + void reconnectWaitsAtLeastOneSecondWithASmallMaxReconnectDelay() throws Exception { + // maxReconnectDelay below the minimum delay must not make a failing message reconnect in a tight loop + String id = UUID.randomUUID().toString(); + String topic = "manualack/smallmax/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60&maxReconnectDelay=0" + + "&clientId=manual-ack-smallmax-" + id; + CamelContext context = createContext(uri, 2); + try { + publish(topic, "fail-small"); + + await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("fail-small")); + List<Long> times = attemptTimes.get("fail-small"); + assertThat(times).hasSize(3); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(1) - times.get(0))).isGreaterThanOrEqualTo(900); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(2) - times.get(1))).isGreaterThanOrEqualTo(900); + } finally { + context.stop(); + } + } + + private static String brokerUrl() { + return "tcp://localhost:" + service.brokerPort(); + } + + /** + * A route that fails the first {@code failures} attempts of every message whose body starts with "fail". + */ + private CamelContext createContext(String uri, int failures) throws Exception { + CamelContext context = new DefaultCamelContext(); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from(uri).process(exchange -> { + String body = exchange.getMessage().getBody(String.class); + int attempt = attempts.computeIfAbsent(body, b -> new AtomicInteger()).incrementAndGet(); + attemptTimes.computeIfAbsent(body, b -> new CopyOnWriteArrayList<>()).add(System.nanoTime()); + if (body.startsWith("fail") && attempt <= failures) { + throw new IllegalStateException("Forced failure of " + body); + } + routed.add(body); + }); + } + }); + context.start(); + return context; + } + + private static void publish(String topic, String... bodies) throws Exception { + publish(topic, 1, false, bodies); + } + + /** + * Publishes the given bodies; a retained publish without bodies clears the retained message of the topic. + */ + private static void publish(String topic, int qos, boolean retained, String... bodies) throws Exception { + MqttClient publisher + = new MqttClient(brokerUrl(), "manual-ack-publisher-" + UUID.randomUUID(), new MemoryPersistence()); + try { + publisher.connect(); + if (retained && bodies.length == 0) { + publisher.publish(topic, new byte[0], qos, true); + } + for (String body : bodies) { + publisher.publish(topic, body.getBytes(StandardCharsets.UTF_8), qos, retained); + } + publisher.disconnect(); + } finally { + publisher.close(); + } + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index cd996a55a4fe..815dbc57e3dd 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -2070,3 +2070,19 @@ now refuses to send a request path, query or header value that contains a contro framing, are still allowed. Routes that put a percent-encoded control character (such as `%0D%0A`) into the `CamelHttpPath` or `CamelHttpQuery` header, which the producer decodes before building the request, are the ones most likely to be affected. + +=== camel-paho-mqtt5 - a failed message with manualAcksEnabled is delivered again or acknowledged + +With `manualAcksEnabled=true`, a message whose exchange failed was left unacknowledged. For a QoS 1 message the +broker never sent it again during the connection, and after `receiveMaximum` such messages it sent the consumer +nothing more (the client completes the QoS 2 flow on its own, so QoS 2 messages did not stop the consumer). Now, for a +QoS 1 message on a session that the broker keeps after the disconnect (`cleanStart=false` and `sessionExpiryInterval` +greater than 0), the consumer connects again after a failed exchange (after one second, doubled for each failed +exchange in a row up to `maxReconnectDelay`, and back to one second after any exchange that succeeds; logged at WARN), +and the broker sends the message again. In all other +cases (a clean session, which is the default, a QoS 2 message, or `cleanStart=false` without `sessionExpiryInterval`, +where the session ends with the connection) the broker never sends the message again, so the failed message is +acknowledged and the failure is left to the Camel error handler. A consumer that uses a client set with the `client` +option leaves a failed QoS 1 message on a session that the broker keeps unacknowledged, as before; as the consumer does +not connect that client, set the `cleanStart` and `sessionExpiryInterval` of the endpoint to match its connection +options. diff --git a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PahoMqtt5ComponentBuilderFactory.java b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PahoMqtt5ComponentBuilderFactory.java index 9852bfca4613..b5b25b0c1423 100644 --- a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PahoMqtt5ComponentBuilderFactory.java +++ b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PahoMqtt5ComponentBuilderFactory.java @@ -498,9 +498,14 @@ public interface PahoMqtt5ComponentBuilderFactory { * default, this is false and message will be automatically acknowledged * upon received by Camel. If set to true, the acknowledgement is * deferred to be acknowledged by Camel at the end of processing the - * message. This ensures that only successfully processed messages is - * acknowledged, and allows to rollback and retry the message in case of - * an error during routing in Camel. + * message. This ensures that only successfully processed messages are + * acknowledged. When the processing of a QoS 1 message fails on a + * session that the broker keeps (cleanStart=false and + * sessionExpiryInterval greater than 0), the consumer connects again so + * that the broker sends the message again. In all other cases the + * broker would never send it again (the client completes the QoS 2 flow + * on its own, and without sessionExpiryInterval the session ends with + * the connection), so the message is acknowledged. * * The option is a: <code>boolean</code> type. * diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PahoMqtt5EndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PahoMqtt5EndpointBuilderFactory.java index 69ea705e4625..753322240118 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PahoMqtt5EndpointBuilderFactory.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PahoMqtt5EndpointBuilderFactory.java @@ -666,9 +666,14 @@ public interface PahoMqtt5EndpointBuilderFactory { * default, this is false and message will be automatically acknowledged * upon received by Camel. If set to true, the acknowledgement is * deferred to be acknowledged by Camel at the end of processing the - * message. This ensures that only successfully processed messages is - * acknowledged, and allows to rollback and retry the message in case of - * an error during routing in Camel. + * message. This ensures that only successfully processed messages are + * acknowledged. When the processing of a QoS 1 message fails on a + * session that the broker keeps (cleanStart=false and + * sessionExpiryInterval greater than 0), the consumer connects again so + * that the broker sends the message again. In all other cases the + * broker would never send it again (the client completes the QoS 2 flow + * on its own, and without sessionExpiryInterval the session ends with + * the connection), so the message is acknowledged. * * The option is a: <code>boolean</code> type. * @@ -687,9 +692,14 @@ public interface PahoMqtt5EndpointBuilderFactory { * default, this is false and message will be automatically acknowledged * upon received by Camel. If set to true, the acknowledgement is * deferred to be acknowledged by Camel at the end of processing the - * message. This ensures that only successfully processed messages is - * acknowledged, and allows to rollback and retry the message in case of - * an error during routing in Camel. + * message. This ensures that only successfully processed messages are + * acknowledged. When the processing of a QoS 1 message fails on a + * session that the broker keeps (cleanStart=false and + * sessionExpiryInterval greater than 0), the consumer connects again so + * that the broker sends the message again. In all other cases the + * broker would never send it again (the client completes the QoS 2 flow + * on its own, and without sessionExpiryInterval the session ends with + * the connection), so the message is acknowledged. * * The option will be converted to a <code>boolean</code> type. *
