This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new c558c2b4cf4a [backport camel-4.22.x] CAMEL-25793: camel-paho-mqtt5 - a
failed message with manualAcksEnabled no longer blocks the consumer (#28268)
c558c2b4cf4a is described below
commit c558c2b4cf4ad16f2a84794efc0cd8a214b221b1
Author: Guillaume Nodet <[email protected]>
AuthorDate: Thu Oct 22 09:23:26 2026 +0200
[backport camel-4.22.x] CAMEL-25793: camel-paho-mqtt5 - a failed message
with manualAcksEnabled no longer blocks the consumer (#28268)
---
.../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.
*
