From 7fa05fe2e231722a9c54081f941c08c37fc14a5b Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Mon, 19 Oct 2026 14:05:46 +0530 Subject: [PATCH 1/6] CAMEL-25793: camel-paho-mqtt5 - a failed message with manualAcksEnabled no longer blocks the consumer With manualAcksEnabled=true a message whose exchange failed was left unacknowledged. MQTT has no negative acknowledgement: the broker sends an unacknowledged message again only on a new connection to the same session, and it sends no more than receiveMaximum unacknowledged messages to the client. So the failed message was never retried as the option promises, and after receiveMaximum failed exchanges the broker sent the consumer nothing more, while the route stayed started. Now, with cleanStart=false the consumer connects again after a failed exchange so that the broker sends the message again. The reconnects of failed exchanges in a row back off (1 s, doubled up to maxReconnectDelay, reset by a successful exchange) and are logged at WARN, on one scheduled executor per consumer. With a clean session, where the broker never sends the message again, the failed message is acknowledged. A client set with the client option keeps the old behaviour (WARN). camel-paho (MQTT 3.1.1, deprecated) has the same behaviour and is not changed. Co-Authored-By: Claude Opus 5.5 --- .../camel/catalog/components/paho-mqtt5.json | 4 +- .../catalog/docs/paho-mqtt5-component.adoc | 18 ++ components/camel-paho-mqtt5/pom.xml | 12 ++ .../component/paho/mqtt5/paho-mqtt5.json | 4 +- .../src/main/docs/paho-mqtt5-component.adoc | 18 ++ .../paho/mqtt5/PahoMqtt5Configuration.java | 5 +- .../paho/mqtt5/PahoMqtt5Consumer.java | 130 ++++++++++++- .../mqtt5/PahoMqtt5ManualAcksFailureTest.java | 177 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 10 + .../dsl/PahoMqtt5ComponentBuilderFactory.java | 8 +- .../dsl/PahoMqtt5EndpointBuilderFactory.java | 16 +- 11 files changed, 386 insertions(+), 16 deletions(-) create mode 100644 components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java 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 3959801c5ee4e..9c38a9921b72f 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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, "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 likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "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 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." }, + "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, "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 fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, "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 type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The quality of service to publish the message at (0, 1 or 2)." }, "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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, - "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 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." }, + "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, "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 messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 ad9dafa2730e7..d951cd0b99246 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,24 @@ 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 the same +session, and it does not send more than `receiveMaximum` unacknowledged messages to the client. So when an exchange +fails: + +* with `cleanStart=false` the consumer disconnects and connects again, and the broker sends the message again; +* with a clean session (`cleanStart=true`, the default) 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 the message stays unacknowledged until that client connects again. + +With `cleanStart=false` 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`, and a successful exchange starts again from one second. +Each such reconnect is logged at WARN. The broker then also sends again the other 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 1d87465aff4a1..fac0b61552b2e 100644 --- a/components/camel-paho-mqtt5/pom.xml +++ b/components/camel-paho-mqtt5/pom.xml @@ -65,6 +65,18 @@ ${project.version} test + + org.apache.artemis + artemis-mqtt-protocol + ${activemq-artemis-version} + test + + + org.apache.camel + camel-test-infra-artemis + ${project.version} + test + 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 3959801c5ee4e..9c38a9921b72f 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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, "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 likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "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 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." }, + "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, "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 fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, "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 type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The quality of service to publish the message at (0, 1 or 2)." }, "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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, - "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 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." }, + "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, "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 messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 ad9dafa2730e7..d951cd0b99246 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,24 @@ 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 the same +session, and it does not send more than `receiveMaximum` unacknowledged messages to the client. So when an exchange +fails: + +* with `cleanStart=false` the consumer disconnects and connects again, and the broker sends the message again; +* with a clean session (`cleanStart=true`, the default) 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 the message stays unacknowledged until that client connects again. + +With `cleanStart=false` 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`, and a successful exchange starts again from one second. +Each such reconnect is logged at WARN. The broker then also sends again the other 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 6ef714816a0e4..14e56cac99b5a 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,9 @@ public void setSessionExpiryInterval(long sessionExpiryInterval) { * * 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 fails, with cleanStart=false the + * consumer connects again so that the broker sends the message again; with a clean session the broker would never + * send it again, 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 bfe641e1ba95f..7890ef2e94902 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,10 @@ 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); + // 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); @@ -134,6 +141,13 @@ public void deliveryComplete(IMqttToken token) { LOG.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic()); client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + + failedMessageReconnects.set(0); + if (stopClient && getEndpoint().getConfiguration().isManualAcksEnabled() + && !getEndpoint().getConfiguration().isCleanStart()) { + reconnectExecutor = getEndpoint().getCamelContext().getExecutorServiceManager() + .newSingleThreadScheduledExecutor(this, "PahoMqtt5-Reconnect"); + } } catch (Exception startException) { MqttClient ownedClient = stopClient ? client : null; if (ownedClient != null) { @@ -181,6 +195,12 @@ private void restartRouteAsync() { protected void doStop() throws Exception { MqttClient ownedClient = stopClient ? client : null; Exception stopException = null; + ScheduledExecutorService executor = reconnectExecutor; + if (executor != null) { + reconnectExecutor = null; + getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor); + } + reconnecting.set(false); try { super.doStop(); @@ -230,6 +250,113 @@ private Exception closeOwnedClient(MqttClient ownedClient, Exception primaryExce return primaryException; } + /** + * MQTT has no negative acknowledgement: the broker sends an unacknowledged message again only when the client + * connects again to the same session, and it sends no more than receiveMaximum unacknowledged messages to the + * client. A failed message that is just left unacknowledged is therefore never retried, and enough of them stop the + * consumer for good. + *

+ * With cleanStart=false 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. With a clean session the broker never sends it 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 (getEndpoint().getConfiguration().isCleanStart()) { + LOG.debug("Acknowledging failed MQTT message with ID: {} of Exchange ID: {} as a clean session never" + + " delivers it again", + mqttMessage.getId(), 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. + */ + 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.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()) { + current.connect(connectionOptions); + current.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } + } catch (Exception e) { + 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(); @@ -249,6 +376,7 @@ public Exchange createExchange(MqttMessage mqttMessage, String topic) { 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) { @@ -260,7 +388,7 @@ public void onComplete(Exchange exchange) { @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 0000000000000..1ead7ee567b9e --- /dev/null +++ b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java @@ -0,0 +1,177 @@ +/* + * 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.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). + */ +public class PahoMqtt5ManualAcksFailureTest { + + @RegisterExtension + static ArtemisService service = ArtemisServiceFactory.createSingletonMQTTService(); + + private final List routed = new CopyOnWriteArrayList<>(); + private final Map attempts = new ConcurrentHashMap<>(); + private final Map> attemptTimes = new ConcurrentHashMap<>(); + + @Test + public 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 + public 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 + public 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 + public 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 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(); + } + } + + 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 { + MqttClient publisher + = new MqttClient(brokerUrl(), "manual-ack-publisher-" + UUID.randomUUID(), new MemoryPersistence()); + try { + publisher.connect(); + for (String body : bodies) { + publisher.publish(topic, body.getBytes(StandardCharsets.UTF_8), 1, false); + } + publisher.disconnect(); + } finally { + publisher.close(); + } + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 5b1e7e99e39d9..e7f7ba4a4c0c8 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -805,6 +805,16 @@ registered. To restore the previous behaviour, set `webhookSecretTokenEnabled=false`: the webhook is then registered without a secret token, and webhook requests are not verified. +=== camel-paho-mqtt5 - a failed message with manualAcksEnabled is delivered again or acknowledged + +With `manualAcksEnabled=true`, a message whose exchange failed was left unacknowledged. The broker never sent it +again during the connection, and after `receiveMaximum` such messages it sent the consumer nothing more. Now, with +`cleanStart=false` the consumer connects again after a failed exchange (after one second, doubled for each failed +exchange in a row up to `maxReconnectDelay`, logged at WARN), and the broker sends the message again; with a +clean session (`cleanStart=true`, the default), where the broker never sends it again, 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 keeps the old behaviour. + === camel-iggy - messageExpiry is sent to Iggy in microseconds The `messageExpiry` option is in seconds, but its value was passed unchanged to Iggy, which reads the message expiry 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 9852bfca4613b..28674da43901f 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,11 @@ default PahoMqtt5ComponentBuilder bridgeErrorHandler(boolean bridgeErrorHandler) * 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 fails, with cleanStart=false the + * consumer connects again so that the broker sends the message again; + * with a clean session the broker would never send it again, 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 69ea705e46257..91c41a195e99d 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,11 @@ default PahoMqtt5EndpointConsumerBuilder willTopic(String willTopic) { * 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 fails, with cleanStart=false the + * consumer connects again so that the broker sends the message again; + * with a clean session the broker would never send it again, so the + * message is acknowledged. * * The option is a: boolean type. * @@ -687,9 +689,11 @@ default PahoMqtt5EndpointConsumerBuilder manualAcksEnabled(boolean manualAcksEna * 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 fails, with cleanStart=false the + * consumer connects again so that the broker sends the message again; + * with a clean session the broker would never send it again, so the + * message is acknowledged. * * The option will be converted to a boolean type. * From a728df13b8b48a94cc56b8bc2ad28b0d84a88e20 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:03:17 +0530 Subject: [PATCH 2/6] CAMEL-25793: camel-paho-mqtt5 - address review: reconnect only for a QoS 1 message on a kept session Paho 1.2.5 holds back only the PUBACK of a QoS 1 message with manual acks; it sends PUBREC as soon as a QoS 2 message is queued for the callback and PUBCOMP as soon as PUBREL arrives, so the broker never sends a QoS 2 message again. Without sessionExpiryInterval (not sent, which means 0) the session ends with the connection. A failed message now makes the consumer reconnect only when it is a QoS 1 message and the broker keeps the session (cleanStart=false and sessionExpiryInterval > 0); in all other cases it is acknowledged, as with a clean session. The reconnect executor is created before the connect, as the broker sends the unacknowledged messages of a kept session right after CONNACK. A reconnect subscribes again only when the broker did not keep the session (connectWithResult/getSessionPresent), so that the retained message of the topic is not sent again. Docs, upgrade guide and the manualAcksEnabled description (component JSON, catalog, DSL) corrected. Co-Authored-By: Claude Opus 5.5 --- .../camel/catalog/components/paho-mqtt5.json | 4 +- .../catalog/docs/paho-mqtt5-component.adoc | 33 ++++--- .../component/paho/mqtt5/paho-mqtt5.json | 4 +- .../src/main/docs/paho-mqtt5-component.adoc | 33 ++++--- .../paho/mqtt5/PahoMqtt5Configuration.java | 8 +- .../paho/mqtt5/PahoMqtt5Consumer.java | 75 ++++++++++----- .../mqtt5/PahoMqtt5ManualAcksFailureTest.java | 94 ++++++++++++++++++- .../pages/camel-4x-upgrade-guide-4_24.adoc | 15 +-- .../dsl/PahoMqtt5ComponentBuilderFactory.java | 11 ++- .../dsl/PahoMqtt5EndpointBuilderFactory.java | 22 +++-- 10 files changed, 220 insertions(+), 79 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 9c38a9921b72f..0f3147d6b6422 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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, "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 likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, + "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 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 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." }, "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 fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, "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 type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The quality of service to publish the message at (0, 1 or 2)." }, "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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, - "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, + "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 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 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." }, "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 messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 d951cd0b99246..959f629be94ed 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 @@ -82,20 +82,25 @@ 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 the same -session, and it does not send more than `receiveMaximum` unacknowledged messages to the client. So when an exchange -fails: - -* with `cleanStart=false` the consumer disconnects and connects again, and the broker sends the message again; -* with a clean session (`cleanStart=true`, the default) 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 the message stays unacknowledged until that client connects again. - -With `cleanStart=false` 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`, and a successful exchange starts again from one second. -Each such reconnect is logged at WARN. The broker then also sends again the other 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. +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 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`, and a successful exchange starts again from one 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/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 9c38a9921b72f..0f3147d6b6422 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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, "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 likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, + "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 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 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." }, "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 fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, "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 type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The quality of service to publish the message at (0, 1 or 2)." }, "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 unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. Whether or not the message should be retained." }, "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 its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to." }, - "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 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 are acknowledged. When the processing fails, with cleanStart=false the consumer connects again so that the broker sends the message again; with a clean session the broker would never send it again, so the message is acknowledged." }, + "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 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 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." }, "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 messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, "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 d951cd0b99246..959f629be94ed 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 @@ -82,20 +82,25 @@ 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 the same -session, and it does not send more than `receiveMaximum` unacknowledged messages to the client. So when an exchange -fails: - -* with `cleanStart=false` the consumer disconnects and connects again, and the broker sends the message again; -* with a clean session (`cleanStart=true`, the default) 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 the message stays unacknowledged until that client connects again. - -With `cleanStart=false` 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`, and a successful exchange starts again from one second. -Each such reconnect is logged at WARN. The broker then also sends again the other 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. +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 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`, and a successful exchange starts again from one 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 14e56cac99b5a..32da4687322ba 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,9 +555,11 @@ public void setSessionExpiryInterval(long sessionExpiryInterval) { * * 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 are acknowledged. When the processing fails, with cleanStart=false the - * consumer connects again so that the broker sends the message again; with a clean session the broker would never - * send it again, so the message is acknowledged. + * 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 7890ef2e94902..5457751212004 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 @@ -73,6 +73,14 @@ protected void doStart() throws Exception { try { connectionOptions = getEndpoint().createMqttConnectionOptions(); + failedMessageReconnects.set(0); + if (stopClient && 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"); + } + if (stopClient) { clientId = getEndpoint().getConfiguration().getClientId(); if (clientId == null) { @@ -141,14 +149,8 @@ public void deliveryComplete(IMqttToken token) { LOG.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic()); client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); - - failedMessageReconnects.set(0); - if (stopClient && getEndpoint().getConfiguration().isManualAcksEnabled() - && !getEndpoint().getConfiguration().isCleanStart()) { - reconnectExecutor = getEndpoint().getCamelContext().getExecutorServiceManager() - .newSingleThreadScheduledExecutor(this, "PahoMqtt5-Reconnect"); - } } catch (Exception startException) { + shutdownReconnectExecutor(); MqttClient ownedClient = stopClient ? client : null; if (ownedClient != null) { client = null; @@ -195,11 +197,7 @@ private void restartRouteAsync() { protected void doStop() throws Exception { MqttClient ownedClient = stopClient ? client : null; Exception stopException = null; - ScheduledExecutorService executor = reconnectExecutor; - if (executor != null) { - reconnectExecutor = null; - getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor); - } + shutdownReconnectExecutor(); reconnecting.set(false); try { super.doStop(); @@ -231,6 +229,24 @@ protected void doStop() throws Exception { } } + 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(), @@ -252,24 +268,27 @@ private Exception closeOwnedClient(MqttClient ownedClient, Exception primaryExce /** * MQTT has no negative acknowledgement: the broker sends an unacknowledged message again only when the client - * connects again to the same session, and it sends no more than receiveMaximum unacknowledged messages to the - * client. A failed message that is just left unacknowledged is therefore never retried, and enough of them stop the - * consumer for good. + * 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. *

- * With cleanStart=false 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. With a clean session the broker never sends it again, so it is acknowledged, which - * frees its place in the receive window. + * 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 (getEndpoint().getConfiguration().isCleanStart()) { - LOG.debug("Acknowledging failed MQTT message with ID: {} of Exchange ID: {} as a clean session never" - + " delivers it again", - mqttMessage.getId(), exchange.getExchangeId()); + 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) { @@ -319,8 +338,14 @@ private void reconnect(MqttClient current) { reconnecting.set(false); try { if (isRunAllowed() && current == client && !current.isConnected()) { - current.connect(connectionOptions); - current.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + 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.debug("The broker did not keep the session of client: {}, subscribing again to topic: {}", + clientId, getEndpoint().getTopic()); + current.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } } } catch (Exception e) { getExceptionHandler().handleException("Failed to reconnect after a failed exchange", e); 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 index 1ead7ee567b9e..3c90d0c2feb93 100644 --- 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 @@ -107,6 +107,88 @@ public void failedMessageIsDeliveredAgainWithPersistentSession() throws Exceptio } } + @Test + public 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 + public 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 + public 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 public void reconnectsAfterFailuresInARowBackOff() throws Exception { // a message that fails three times in a row: the reconnects wait 1 s, then 2 s capped to maxReconnectDelay @@ -162,12 +244,22 @@ public void configure() { } 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), 1, false); + publisher.publish(topic, body.getBytes(StandardCharsets.UTF_8), qos, retained); } publisher.disconnect(); } finally { diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index e7f7ba4a4c0c8..e0142a21fc1c1 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -807,13 +807,16 @@ secret token, and webhook requests are not verified. === camel-paho-mqtt5 - a failed message with manualAcksEnabled is delivered again or acknowledged -With `manualAcksEnabled=true`, a message whose exchange failed was left unacknowledged. The broker never sent it -again during the connection, and after `receiveMaximum` such messages it sent the consumer nothing more. Now, with -`cleanStart=false` the consumer connects again after a failed exchange (after one second, doubled for each failed -exchange in a row up to `maxReconnectDelay`, logged at WARN), and the broker sends the message again; with a -clean session (`cleanStart=true`, the default), where the broker never sends it again, the failed message is +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`, 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 keeps the old behaviour. +option leaves a failed QoS 1 message on a session that the broker keeps unacknowledged, as before. === camel-iggy - messageExpiry is sent to Iggy in microseconds 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 28674da43901f..2ea6a4869dade 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 @@ -499,10 +499,13 @@ default PahoMqtt5ComponentBuilder bridgeErrorHandler(boolean bridgeErrorHandler) * 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 are - * acknowledged. When the processing fails, with cleanStart=false the - * consumer connects again so that the broker sends the message again; - * with a clean session the broker would never send it again, so the - * message is acknowledged. + * 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 91c41a195e99d..8ae4d12feaedf 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 @@ -667,10 +667,13 @@ default PahoMqtt5EndpointConsumerBuilder willTopic(String willTopic) { * 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 are - * acknowledged. When the processing fails, with cleanStart=false the - * consumer connects again so that the broker sends the message again; - * with a clean session the broker would never send it again, so the - * message is acknowledged. + * 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: boolean type. * @@ -690,10 +693,13 @@ default PahoMqtt5EndpointConsumerBuilder manualAcksEnabled(boolean manualAcksEna * 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 are - * acknowledged. When the processing fails, with cleanStart=false the - * consumer connects again so that the broker sends the message again; - * with a clean session the broker would never send it again, so the - * message is acknowledged. + * 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 boolean type. * From a251659ceb0c65ff983a5e06d4ff0056373d7e1f Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:16:34 +0530 Subject: [PATCH 3/6] CAMEL-25793: camel-paho-mqtt5 - address review: wrap the manualAcksEnabled DSL javadoc as the generator does The endpoint and component DSL javadoc of the manualAcksEnabled description was wrapped by hand one column too wide; CI regenerates it with lines of at most 80 columns. Co-Authored-By: Claude Opus 5.5 --- .../dsl/PahoMqtt5ComponentBuilderFactory.java | 8 ++++---- .../dsl/PahoMqtt5EndpointBuilderFactory.java | 16 ++++++++-------- 2 files changed, 12 insertions(+), 12 deletions(-) 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 2ea6a4869dade..b5b25b0c1423a 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 @@ -502,10 +502,10 @@ default PahoMqtt5ComponentBuilder bridgeErrorHandler(boolean bridgeErrorHandler) * 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. + * 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 8ae4d12feaedf..7533222401189 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 @@ -670,10 +670,10 @@ default PahoMqtt5EndpointConsumerBuilder willTopic(String willTopic) { * 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. + * 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: boolean type. * @@ -696,10 +696,10 @@ default PahoMqtt5EndpointConsumerBuilder manualAcksEnabled(boolean manualAcksEna * 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. + * 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 boolean type. * From a1664124d94730a4338066d2770834aefe628b67 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 12:27:41 +0530 Subject: [PATCH 4/6] CAMEL-25793: camel-paho-mqtt5 - address review: reconnect after at least the minimum delay, client option docs - A maxReconnectDelay below the minimum reconnect delay (1 second) made a message that always fails reconnect in a tight loop: the delay is now never below the minimum. - The docs say that with a client set by the client option the consumer decides with the cleanStart and sessionExpiryInterval of the endpoint, which must match the connection options of that client. Co-Authored-By: Claude Opus 5.5 --- .../catalog/docs/paho-mqtt5-component.adoc | 12 +++++----- .../src/main/docs/paho-mqtt5-component.adoc | 12 +++++----- .../paho/mqtt5/PahoMqtt5Consumer.java | 5 +++-- .../mqtt5/PahoMqtt5ManualAcksFailureTest.java | 22 +++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 4 +++- 5 files changed, 42 insertions(+), 13 deletions(-) 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 959f629be94ed..3c8098b6aac44 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 @@ -94,13 +94,15 @@ arrives, so the broker never sends a QoS 2 message again. So when an exchange fa `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. + 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`, and a successful exchange starts again from one 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. +long, up to `maxReconnectDelay` (but at least one second), and a successful exchange starts again from one 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/docs/paho-mqtt5-component.adoc b/components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc index 959f629be94ed..3c8098b6aac44 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 @@ -94,13 +94,15 @@ arrives, so the broker never sends a QoS 2 message again. So when an exchange fa `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. + 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`, and a successful exchange starts again from one 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. +long, up to `maxReconnectDelay` (but at least one second), and a successful exchange starts again from one 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/PahoMqtt5Consumer.java b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java index 5457751212004..220efdbbae122 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 @@ -314,14 +314,15 @@ private void reconnectAfterFailure(MqttClient current, Exchange exchange, MqttMe /** * 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. + * 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.min(delay, max); + return Math.max(min, Math.min(delay, max)); } private void reconnect(MqttClient current) { 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 index 3c90d0c2feb93..d4dcc4ca1fdfe 100644 --- 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 @@ -216,6 +216,28 @@ public void reconnectsAfterFailuresInARowBackOff() throws Exception { } } + @Test + public 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 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(); } diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index e0142a21fc1c1..0aace241d71c4 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -816,7 +816,9 @@ exchange in a row up to `maxReconnectDelay`, logged at WARN), and the broker sen 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. +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. === camel-iggy - messageExpiry is sent to Iggy in microseconds From 99be40e84a8bb4341736ad088b4ee5f85945085a Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 16:56:21 +0530 Subject: [PATCH 5/6] CAMEL-25793: camel-paho-mqtt5 - address review: no subscribe on our own reconnect after a lost connection, WARN without session Paho 1.2.5 sets MqttAsyncClient.reconnecting when the client loses its connection with automaticReconnect=true and never clears it, so every later connect ends in connectComplete(true, ...). Once the client had lost a connection, the reconnect after a failed exchange therefore made the connectComplete handler subscribe again, and the broker sent the retained message of the topic again (and the subscribe ran twice when the session was gone). reconnect() now sets a flag that connectComplete checks and clears; the handler subscribes again only for the automatic reconnects of Paho. When the reconnect for a redelivery finds no session, the failed message and the other unacknowledged ones are gone: this is now logged at WARN. The docs and the upgrade guide note say that any exchange that succeeds resets the backoff, so with other messages succeeding in between a message that always fails reconnects about once a second. The test class and its methods are package-private. Co-Authored-By: Claude Opus 5.5 --- .../catalog/docs/paho-mqtt5-component.adoc | 4 +- .../src/main/docs/paho-mqtt5-component.adoc | 4 +- .../paho/mqtt5/PahoMqtt5Consumer.java | 15 ++++- .../mqtt5/PahoMqtt5ManualAcksFailureTest.java | 58 ++++++++++++++++--- .../pages/camel-4x-upgrade-guide-4_24.adoc | 3 +- 5 files changed, 70 insertions(+), 14 deletions(-) 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 3c8098b6aac44..4ebe6c0cc4d25 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 @@ -99,7 +99,9 @@ arrives, so the broker never sends a QoS 2 message again. So when an exchange fa 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 a successful exchange starts again from one second. +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. 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 3c8098b6aac44..4ebe6c0cc4d25 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 @@ -99,7 +99,9 @@ arrives, so the broker never sends a QoS 2 message again. So when an exchange fa 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 a successful exchange starts again from one second. +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. 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 220efdbbae122..64486647169b9 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 @@ -49,6 +49,10 @@ public class PahoMqtt5Consumer extends DefaultConsumer { 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; @@ -98,7 +102,9 @@ protected void doStart() throws Exception { @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) { @@ -199,6 +205,7 @@ protected void doStop() throws Exception { Exception stopException = null; shutdownReconnectExecutor(); reconnecting.set(false); + ownReconnect.set(false); try { super.doStop(); @@ -339,16 +346,20 @@ private void reconnect(MqttClient current) { 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.debug("The broker did not keep the session of client: {}, subscribing again to topic: {}", + 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"); 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 index d4dcc4ca1fdfe..9cd0cde4bfe94 100644 --- 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 @@ -31,6 +31,7 @@ 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; @@ -43,7 +44,7 @@ * 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). */ -public class PahoMqtt5ManualAcksFailureTest { +class PahoMqtt5ManualAcksFailureTest { @RegisterExtension static ArtemisService service = ArtemisServiceFactory.createSingletonMQTTService(); @@ -53,7 +54,7 @@ public class PahoMqtt5ManualAcksFailureTest { private final Map> attemptTimes = new ConcurrentHashMap<>(); @Test - public void failedMessagesDoNotStopTheConsumer() throws Exception { + 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(); @@ -72,7 +73,7 @@ public void failedMessagesDoNotStopTheConsumer() throws Exception { } @Test - public void acknowledgedMessagesKeepTheReceiveWindowOpen() throws Exception { + void acknowledgedMessagesKeepTheReceiveWindowOpen() throws Exception { String id = UUID.randomUUID().toString(); String topic = "manualack/ok/" + id; String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() @@ -88,7 +89,7 @@ public void acknowledgedMessagesKeepTheReceiveWindowOpen() throws Exception { } @Test - public void failedMessageIsDeliveredAgainWithPersistentSession() throws Exception { + 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; @@ -108,7 +109,7 @@ public void failedMessageIsDeliveredAgainWithPersistentSession() throws Exceptio } @Test - public void failedQos2MessageIsAcknowledgedWithoutReconnect() throws Exception { + 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(); @@ -138,7 +139,7 @@ public void failedQos2MessageIsAcknowledgedWithoutReconnect() throws Exception { } @Test - public void failedMessageIsAcknowledgedWhenTheSessionEndsWithTheConnection() throws Exception { + 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(); @@ -164,7 +165,7 @@ public void failedMessageIsAcknowledgedWhenTheSessionEndsWithTheConnection() thr } @Test - public void reconnectToKeptSessionDoesNotSubscribeAgain() throws Exception { + 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(); @@ -190,7 +191,46 @@ public void reconnectToKeptSessionDoesNotSubscribeAgain() throws Exception { } @Test - public void reconnectsAfterFailuresInARowBackOff() throws Exception { + 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, which may send the retained message again + await().pollDelay(1500, TimeUnit.MILLISECONDS).atMost(3, TimeUnit.SECONDS).until(() -> true); + 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; @@ -217,7 +257,7 @@ public void reconnectsAfterFailuresInARowBackOff() throws Exception { } @Test - public void reconnectWaitsAtLeastOneSecondWithASmallMaxReconnectDelay() throws Exception { + 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; diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 0aace241d71c4..8ced53d334ed8 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -812,7 +812,8 @@ broker never sent it again during the connection, and after `receiveMaximum` suc 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`, logged at WARN), and the broker sends the message again. In all other +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` From 4edb388ba0e14521126c18bafaa0decf106c81b5 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 20:56:50 +0530 Subject: [PATCH 6/6] CAMEL-25793: camel-paho-mqtt5 - address review: wait for the retained resend instead of a fixed delay Paho's automatic re-subscribe uses Retain Handling 0, so the broker always sends the retained message again; the test now waits for that second delivery before it takes the count. Co-Authored-By: Claude Opus 5.5 --- .../component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 index 9cd0cde4bfe94..64f18ae24ce95 100644 --- 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 @@ -215,8 +215,9 @@ void reconnectAfterAnEarlierConnectionLossDoesNotSubscribeAgain() throws Excepti other.close(); } await().atMost(10, TimeUnit.SECONDS).until(() -> consumer.getClient().isConnected()); - // the automatic reconnect of Paho subscribes again, which may send the retained message again - await().pollDelay(1500, TimeUnit.MILLISECONDS).atMost(3, TimeUnit.SECONDS).until(() -> true); + // 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");