CAMEL-25793: camel-paho-mqtt5 - a failed message with manualAcksEnabled no longer blocks the consumer

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 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 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 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." },

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/paho-mqtt5-component.adoc

@@ -79,6 +79,33 @@ String payload = "message";
 producerTemplate.sendBody("paho-mqtt5:topic", payload);
 ----
 
+=== Manual acknowledgements
+
+With `manualAcksEnabled=true` a message is acknowledged when its exchange completes. MQTT has no negative
+acknowledgement: the broker sends an unacknowledged message again only when the client connects again to a session
+that the broker kept, and it does not send more than `receiveMaximum` unacknowledged messages to the client. The
+client holds back only the acknowledgement of a QoS 1 message; it completes the QoS 2 flow on its own when the message
+arrives, so the broker never sends a QoS 2 message again. So when an exchange fails:
+
+* for a QoS 1 message on a session that the broker keeps after the disconnect (`cleanStart=false` and
+  `sessionExpiryInterval` greater than 0) the consumer disconnects and connects again, and the broker sends the
+  message again;
+* in all other cases (a clean session, which is the default, a QoS 2 message, or `cleanStart=false` without
+  `sessionExpiryInterval`, where the session ends with the connection) the broker never sends the message again, so
+  the consumer acknowledges it, and the failure is left to the Camel error handler;
+* with a client set by the `client` option, a QoS 1 message on a session that the broker keeps stays unacknowledged
+  until that client connects again. The consumer does not connect that client, so it tells the cases apart with the
+  `cleanStart` and `sessionExpiryInterval` of the endpoint: set them to match the connection options of the client.
+  With the defaults (a clean session) a failed message is acknowledged.
+
+The reconnects of failed exchanges in a row back off: the first one waits one second, each further one waits twice as
+long, up to `maxReconnectDelay` (but at least one second), and any exchange that succeeds, also one of another
+message, starts again from one second. So on a topic where other messages succeed in between, a message that always
+fails makes the consumer reconnect about once a second.
+Each such reconnect is logged at WARN. The broker then also sends again the other QoS 1 messages that were not
+acknowledged yet, such as the ones still being routed. A message that always fails is still delivered again and again;
+use a Camel error handler, for example a dead letter channel, for such messages.
+
 
 == Examples
 

components/camel-paho-mqtt5/pom.xml

@@ -65,6 +65,18 @@
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.artemis</groupId>
+            <artifactId>artemis-mqtt-protocol</artifactId>
+            <version>${activemq-artemis-version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-test-infra-artemis</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>