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..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 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." }, 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..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 @@ -79,6 +79,33 @@ String payload = "message"; producerTemplate.sendBody("paho-mqtt5:topic", payload); ---- +=== Manual acknowledgements + +With `manualAcksEnabled=true` a message is acknowledged when its exchange completes. MQTT has no negative +acknowledgement: the broker sends an unacknowledged message again only when the client connects again to a session +that the broker kept, and it does not send more than `receiveMaximum` unacknowledged messages to the client. The +client holds back only the acknowledgement of a QoS 1 message; it completes the QoS 2 flow on its own when the message +arrives, so the broker never sends a QoS 2 message again. So when an exchange fails: + +* for a QoS 1 message on a session that the broker keeps after the disconnect (`cleanStart=false` and + `sessionExpiryInterval` greater than 0) the consumer disconnects and connects again, and the broker sends the + message again; +* in all other cases (a clean session, which is the default, a QoS 2 message, or `cleanStart=false` without + `sessionExpiryInterval`, where the session ends with the connection) the broker never sends the message again, so + the consumer acknowledges it, and the failure is left to the Camel error handler; +* with a client set by the `client` option, a QoS 1 message on a session that the broker keeps stays unacknowledged + until that client connects again. The consumer does not connect that client, so it tells the cases apart with the + `cleanStart` and `sessionExpiryInterval` of the endpoint: set them to match the connection options of the client. + With the defaults (a clean session) a failed message is acknowledged. + +The reconnects of failed exchanges in a row back off: the first one waits one second, each further one waits twice as +long, up to `maxReconnectDelay` (but at least one second), and any exchange that succeeds, also one of another +message, starts again from one second. So on a topic where other messages succeed in between, a message that always +fails makes the consumer reconnect about once a second. +Each such reconnect is logged at WARN. The broker then also sends again the other QoS 1 messages that were not +acknowledged yet, such as the ones still being routed. A message that always fails is still delivered again and again; +use a Camel error handler, for example a dead letter channel, for such messages. + == Examples diff --git a/components/camel-paho-mqtt5/pom.xml b/components/camel-paho-mqtt5/pom.xml index 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..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 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." }, 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..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 @@ -79,6 +79,33 @@ String payload = "message"; producerTemplate.sendBody("paho-mqtt5:topic", payload); ---- +=== Manual acknowledgements + +With `manualAcksEnabled=true` a message is acknowledged when its exchange completes. MQTT has no negative +acknowledgement: the broker sends an unacknowledged message again only when the client connects again to a session +that the broker kept, and it does not send more than `receiveMaximum` unacknowledged messages to the client. The +client holds back only the acknowledgement of a QoS 1 message; it completes the QoS 2 flow on its own when the message +arrives, so the broker never sends a QoS 2 message again. So when an exchange fails: + +* for a QoS 1 message on a session that the broker keeps after the disconnect (`cleanStart=false` and + `sessionExpiryInterval` greater than 0) the consumer disconnects and connects again, and the broker sends the + message again; +* in all other cases (a clean session, which is the default, a QoS 2 message, or `cleanStart=false` without + `sessionExpiryInterval`, where the session ends with the connection) the broker never sends the message again, so + the consumer acknowledges it, and the failure is left to the Camel error handler; +* with a client set by the `client` option, a QoS 1 message on a session that the broker keeps stays unacknowledged + until that client connects again. The consumer does not connect that client, so it tells the cases apart with the + `cleanStart` and `sessionExpiryInterval` of the endpoint: set them to match the connection options of the client. + With the defaults (a clean session) a failed message is acknowledged. + +The reconnects of failed exchanges in a row back off: the first one waits one second, each further one waits twice as +long, up to `maxReconnectDelay` (but at least one second), and any exchange that succeeds, also one of another +message, starts again from one second. So on a topic where other messages succeed in between, a message that always +fails makes the consumer reconnect about once a second. +Each such reconnect is logged at WARN. The broker then also sends again the other QoS 1 messages that were not +acknowledged yet, such as the ones still being routed. A message that always fails is still delivered again and again; +use a Camel error handler, for example a dead letter channel, for such messages. + == Examples diff --git a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Configuration.java b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Configuration.java index 6ef714816a0e4..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,8 +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 is acknowledged, and allows to rollback and retry the message in case - * of an error during routing in Camel. + * that only successfully processed messages are acknowledged. When the processing of a QoS 1 message fails on a + * session that the broker keeps (cleanStart=false and sessionExpiryInterval greater than 0), the consumer connects + * again so that the broker sends the message again. In all other cases the broker would never send it again (the + * client completes the QoS 2 flow on its own, and without sessionExpiryInterval the session ends with the + * connection), so the message is acknowledged. */ public boolean isManualAcksEnabled() { return this.manualAcksEnabled; diff --git a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java index bfe641e1ba95f..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 @@ -17,7 +17,10 @@ package org.apache.camel.component.paho.mqtt5; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.camel.AsyncCallback; import org.apache.camel.Endpoint; @@ -45,6 +48,14 @@ public class PahoMqtt5Consumer extends DefaultConsumer { private volatile boolean stopClient; private volatile MqttConnectionOptions connectionOptions; private final AtomicBoolean restarting = new AtomicBoolean(false); + private final AtomicBoolean reconnecting = new AtomicBoolean(false); + // set while the consumer connects the client again after a failed exchange: Paho reports that connect to + // connectComplete with reconnect=true once the client lost a connection on its own (it never clears its + // reconnecting flag), and the consumer subscribes again there only for the automatic reconnects of Paho + private final AtomicBoolean ownReconnect = new AtomicBoolean(false); + // failed exchanges in a row that made the consumer reconnect, for the backoff; reset by a successful exchange + private final AtomicInteger failedMessageReconnects = new AtomicInteger(); + private volatile ScheduledExecutorService reconnectExecutor; public PahoMqtt5Consumer(Endpoint endpoint, Processor processor) { super(endpoint, processor); @@ -66,6 +77,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) { @@ -83,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) { @@ -135,6 +156,7 @@ public void deliveryComplete(IMqttToken token) { LOG.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic()); client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); } catch (Exception startException) { + shutdownReconnectExecutor(); MqttClient ownedClient = stopClient ? client : null; if (ownedClient != null) { client = null; @@ -181,6 +203,9 @@ private void restartRouteAsync() { protected void doStop() throws Exception { MqttClient ownedClient = stopClient ? client : null; Exception stopException = null; + shutdownReconnectExecutor(); + reconnecting.set(false); + ownReconnect.set(false); try { super.doStop(); @@ -211,6 +236,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(), @@ -230,6 +273,127 @@ 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 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. + *

+ * For a QoS 1 message on a session that the broker keeps after the disconnect (cleanStart=false and + * sessionExpiryInterval greater than 0) the client reconnects, so that the broker sends the message again; the + * reconnects of failed exchanges in a row back off (see {@link #reconnectDelay(int)}), so that a message that + * always fails does not reconnect in a tight loop. In every other case the broker never sends the message again, so + * it is acknowledged, which frees its place in the receive window. + */ + private void rollback(Exchange exchange, MqttMessage mqttMessage) { + MqttClient current = client; + if (current == null) { + return; + } + if (mqttMessage.getQos() != 1 || !isSessionKeptAfterDisconnect()) { + LOG.debug("Acknowledging failed MQTT message with ID: {} (QoS {}) of Exchange ID: {} as the broker does not" + + " deliver it again", + mqttMessage.getId(), mqttMessage.getQos(), exchange.getExchangeId()); + try { + current.messageArrivedComplete(mqttMessage.getId(), mqttMessage.getQos()); + } catch (MqttException e) { + getExceptionHandler().handleException( + "Error acknowledging failed MQTT message with ID: " + mqttMessage.getId(), exchange, e); + } + } else if (stopClient) { + reconnectAfterFailure(current, exchange, mqttMessage); + } else { + LOG.warn("Rollback due to error processing Exchange ID: {}: MQTT message with ID: {} stays unacknowledged" + + " until the externally provided client connects again", + exchange.getExchangeId(), mqttMessage.getId()); + } + } + + private void reconnectAfterFailure(MqttClient current, Exchange exchange, MqttMessage mqttMessage) { + String reason = "Rollback due to error processing Exchange ID: " + exchange.getExchangeId() + + ", so that the broker delivers MQTT message with ID: " + mqttMessage.getId() + " again"; + if (!scheduleReconnect(current, reason)) { + LOG.debug("{}: a reconnect is already scheduled", reason); + } + } + + /** + * The delay before the reconnect of the given failed exchange in a row: the minimum automatic reconnect delay of + * the client (1 second), doubled for each further failed exchange in a row, up to maxReconnectDelay, but never less + * than the minimum (a maxReconnectDelay below it would reconnect in a tight loop). + */ + private long reconnectDelay(int attempt) { + long max = Math.max(0, getEndpoint().getConfiguration().getMaxReconnectDelay()); + MqttConnectionOptions options = connectionOptions; + long min = options != null ? TimeUnit.SECONDS.toMillis(options.getAutomaticReconnectMinDelay()) : 1000; + long delay = min << Math.min(attempt - 1, 20); + return Math.max(min, Math.min(delay, max)); + } + + private void reconnect(MqttClient current) { + try { + // the message callback may still be running (disconnect is not allowed from it), so this runs here + if (isRunAllowed() && current == client && current.isConnected()) { + current.disconnect(); + } + } catch (Exception e) { + getExceptionHandler().handleException("Failed to disconnect after a failed exchange", e); + } + // the disconnect covers every message that failed until now; a message the broker sends again on the new + // connection can fail before this method returns, and must be able to schedule the next reconnect + reconnecting.set(false); + try { + if (isRunAllowed() && current == client && !current.isConnected()) { + ownReconnect.set(true); + IMqttToken token = current.connectWithResult(connectionOptions); + // the subscription is part of the session; subscribing again would make the broker send the retained + // message of the topic again (Retain Handling 0) + if (!token.getSessionPresent()) { + LOG.warn("The broker did not keep the session of client: {}: the failed MQTT message and the other" + + " unacknowledged ones are not delivered again; subscribing again to topic: {}", + clientId, getEndpoint().getTopic()); + current.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } + } + } catch (Exception e) { + // a failed connect does not call connectComplete + ownReconnect.set(false); + getExceptionHandler().handleException("Failed to reconnect after a failed exchange", e); + // the client does not reconnect on its own after a disconnect it was asked for, so try again later + scheduleReconnect(current, "Reconnect failed"); + } + } + + /** + * Schedules a reconnect after the backoff delay, unless one is already scheduled or the consumer is stopping. + */ + private boolean scheduleReconnect(MqttClient current, String reason) { + ScheduledExecutorService executor = reconnectExecutor; + if (executor == null || !isRunAllowed() || !reconnecting.compareAndSet(false, true)) { + return false; + } + int attempt = failedMessageReconnects.incrementAndGet(); + long delay = reconnectDelay(attempt); + LOG.warn("{}: reconnecting in {} ms (reconnect {} in a row)", reason, delay, attempt); + try { + executor.schedule(() -> reconnect(current), delay, TimeUnit.MILLISECONDS); + return true; + } catch (RuntimeException e) { + // the consumer is stopping + reconnecting.set(false); + return false; + } + } + + /** + * The number of failed exchanges in a row that made the consumer reconnect. Package-private, for tests. + */ + int getFailedMessageReconnects() { + return failedMessageReconnects.get(); + } + @Override public PahoMqtt5Endpoint getEndpoint() { return (PahoMqtt5Endpoint) super.getEndpoint(); @@ -249,6 +413,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 +425,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..64f18ae24ce95 --- /dev/null +++ b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java @@ -0,0 +1,332 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.paho.mqtt5; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.test.infra.artemis.services.ArtemisService; +import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory; +import org.eclipse.paho.mqttv5.client.MqttClient; +import org.eclipse.paho.mqttv5.client.MqttConnectionOptionsBuilder; +import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * With manualAcksEnabled=true a message whose exchange failed is not acknowledged. The broker sends at most + * receiveMaximum unacknowledged QoS 1/2 messages to a client, and sends an unacknowledged message again only when the + * client connects again to the same session. These tests use a real broker (embedded Artemis, MQTT 5). + */ +class PahoMqtt5ManualAcksFailureTest { + + @RegisterExtension + static ArtemisService service = ArtemisServiceFactory.createSingletonMQTTService(); + + private final List routed = new CopyOnWriteArrayList<>(); + private final Map attempts = new ConcurrentHashMap<>(); + private final Map> attemptTimes = new ConcurrentHashMap<>(); + + @Test + void failedMessagesDoNotStopTheConsumer() throws Exception { + // clean session (the default) and a receive maximum of 2: after two failed exchanges the broker has two + // unacknowledged messages in flight for this client + String id = UUID.randomUUID().toString(); + String topic = "manualack/clean/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&receiveMaximum=2&clientId=manual-ack-clean-" + id; + CamelContext context = createContext(uri, 1); + try { + publish(topic, "fail-1", "fail-2", "ok-3", "ok-4"); + + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("ok-3", "ok-4")); + } finally { + context.stop(); + } + } + + @Test + void acknowledgedMessagesKeepTheReceiveWindowOpen() throws Exception { + String id = UUID.randomUUID().toString(); + String topic = "manualack/ok/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&receiveMaximum=2&clientId=manual-ack-ok-" + id; + CamelContext context = createContext(uri, 1); + try { + publish(topic, "ok-1", "ok-2", "ok-3", "ok-4"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("ok-1", "ok-2", "ok-3", "ok-4")); + } finally { + context.stop(); + } + } + + @Test + void failedMessageIsDeliveredAgainWithPersistentSession() throws Exception { + // a session that the broker keeps: the documented rollback and retry of a failed message + String id = UUID.randomUUID().toString(); + String topic = "manualack/persistent/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60" + + "&clientId=manual-ack-persistent-" + id; + CamelContext context = createContext(uri, 1); + try { + publish(topic, "fail-1", "ok-2"); + + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactlyInAnyOrder("fail-1", "ok-2")); + assertThat(attempts.get("fail-1")).hasValue(2); + } finally { + context.stop(); + } + } + + @Test + void failedQos2MessageIsAcknowledgedWithoutReconnect() throws Exception { + // Paho completes the QoS 2 flow (PUBREC, PUBCOMP) without waiting for the manual acknowledgement, so the broker + // never sends a QoS 2 message again, even to a session that it keeps: a reconnect would not retry it + String id = UUID.randomUUID().toString(); + String topic = "manualack/qos2/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=2&manualAcksEnabled=true&receiveMaximum=2&cleanStart=false&sessionExpiryInterval=60" + + "&clientId=manual-ack-qos2-" + id; + CamelContext context = createContext(uri, 1); + try { + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + publish(topic, 2, false, "fail-1", "fail-2"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { + assertThat(attempts.get("fail-1")).hasValue(1); + assertThat(attempts.get("fail-2")).hasValue(1); + }); + await().during(1500, TimeUnit.MILLISECONDS).atMost(3, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(consumer.getFailedMessageReconnects()).isZero()); + + publish(topic, 2, false, "ok-3", "ok-4"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("ok-3", "ok-4")); + assertThat(attempts.get("fail-1")).hasValue(1); + assertThat(attempts.get("fail-2")).hasValue(1); + } finally { + context.stop(); + } + } + + @Test + void failedMessageIsAcknowledgedWhenTheSessionEndsWithTheConnection() throws Exception { + // cleanStart=false without sessionExpiryInterval: the interval is not sent, which means 0, so the broker ends + // the session when the client disconnects, and a reconnect would lose the messages instead of sending them again + String id = UUID.randomUUID().toString(); + String topic = "manualack/noexpiry/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&receiveMaximum=1&cleanStart=false" + + "&clientId=manual-ack-noexpiry-" + id; + CamelContext context = createContext(uri, 1); + try { + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + publish(topic, "fail-1"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(attempts.get("fail-1")).hasValue(1)); + await().during(1500, TimeUnit.MILLISECONDS).atMost(3, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(consumer.getFailedMessageReconnects()).isZero()); + + // with a receive maximum of 1 the broker sends the next message only after fail-1 is acknowledged + publish(topic, "ok-2"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("ok-2")); + assertThat(attempts.get("fail-1")).hasValue(1); + } finally { + context.stop(); + } + } + + @Test + void reconnectToKeptSessionDoesNotSubscribeAgain() throws Exception { + // the subscription is part of the session that the broker kept; subscribing again would make the broker send + // the retained message of the topic again + String id = UUID.randomUUID().toString(); + String topic = "manualack/retained/" + id; + publish(topic, 1, true, "retained-0"); + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60" + + "&clientId=manual-ack-retained-" + id; + CamelContext context = createContext(uri, 1); + try { + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("retained-0")); + + publish(topic, "fail-1"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(routed).containsExactly("retained-0", "fail-1")); + assertThat(attempts.get("fail-1")).hasValue(2); + await().during(1, TimeUnit.SECONDS).atMost(2, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(attempts.get("retained-0")).hasValue(1)); + } finally { + context.stop(); + publish(topic, 1, true); + } + } + + @Test + void reconnectAfterAnEarlierConnectionLossDoesNotSubscribeAgain() throws Exception { + // once the client lost a connection on its own, Paho reports every later connect to connectComplete as a + // reconnect: the reconnect of a failed exchange must still not subscribe again to the session the broker kept + String id = UUID.randomUUID().toString(); + String topic = "manualack/lost/" + id; + String clientId = "manual-ack-lost-" + id; + publish(topic, 1, true, "retained-0"); + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60&clientId=" + clientId; + CamelContext context = createContext(uri, 1); + try { + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("retained-0")); + + // another connection with the same client id takes the session over: the broker disconnects the consumer's + // client, which reconnects on its own (automaticReconnect) once the session is free again + MqttClient other = new MqttClient(brokerUrl(), clientId, new MemoryPersistence()); + try { + other.connect(new MqttConnectionOptionsBuilder().cleanStart(false).sessionExpiryInterval(60L).build()); + other.disconnect(); + } finally { + other.close(); + } + await().atMost(10, TimeUnit.SECONDS).until(() -> consumer.getClient().isConnected()); + // the automatic reconnect of Paho subscribes again (Retain Handling 0), so the broker sends the retained + // message again + await().atMost(10, TimeUnit.SECONDS).until(() -> attempts.get("retained-0").get() == 2); + int retainedBefore = attempts.get("retained-0").get(); + + publish(topic, "fail-1"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(attempts.get("fail-1")).hasValue(2)); + await().during(1, TimeUnit.SECONDS).atMost(2, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(attempts.get("retained-0")).hasValue(retainedBefore)); + } finally { + context.stop(); + publish(topic, 1, true); + } + } + + @Test + void reconnectsAfterFailuresInARowBackOff() throws Exception { + // a message that fails three times in a row: the reconnects wait 1 s, then 2 s capped to maxReconnectDelay + String id = UUID.randomUUID().toString(); + String topic = "manualack/backoff/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60&maxReconnectDelay=1500" + + "&clientId=manual-ack-backoff-" + id; + CamelContext context = createContext(uri, 3); + try { + publish(topic, "fail-3"); + + await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("fail-3")); + List times = attemptTimes.get("fail-3"); + assertThat(times).hasSize(4); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(1) - times.get(0))).isGreaterThanOrEqualTo(900); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(2) - times.get(1))).isGreaterThanOrEqualTo(1400); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(3) - times.get(2))).isGreaterThanOrEqualTo(1400); + // the successful exchange resets the backoff + PahoMqtt5Consumer consumer = (PahoMqtt5Consumer) context.getRoutes().get(0).getConsumer(); + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(consumer.getFailedMessageReconnects()).isZero()); + } finally { + context.stop(); + } + } + + @Test + void reconnectWaitsAtLeastOneSecondWithASmallMaxReconnectDelay() throws Exception { + // maxReconnectDelay below the minimum delay must not make a failing message reconnect in a tight loop + String id = UUID.randomUUID().toString(); + String topic = "manualack/smallmax/" + id; + String uri = "paho-mqtt5:" + topic + "?brokerUrl=" + brokerUrl() + + "&qos=1&manualAcksEnabled=true&cleanStart=false&sessionExpiryInterval=60&maxReconnectDelay=0" + + "&clientId=manual-ack-smallmax-" + id; + CamelContext context = createContext(uri, 2); + try { + publish(topic, "fail-small"); + + await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> assertThat(routed).containsExactly("fail-small")); + List times = attemptTimes.get("fail-small"); + assertThat(times).hasSize(3); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(1) - times.get(0))).isGreaterThanOrEqualTo(900); + assertThat(TimeUnit.NANOSECONDS.toMillis(times.get(2) - times.get(1))).isGreaterThanOrEqualTo(900); + } finally { + context.stop(); + } + } + + private static String brokerUrl() { + return "tcp://localhost:" + service.brokerPort(); + } + + /** + * A route that fails the first {@code failures} attempts of every message whose body starts with "fail". + */ + private CamelContext createContext(String uri, int failures) throws Exception { + CamelContext context = new DefaultCamelContext(); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from(uri).process(exchange -> { + String body = exchange.getMessage().getBody(String.class); + int attempt = attempts.computeIfAbsent(body, b -> new AtomicInteger()).incrementAndGet(); + attemptTimes.computeIfAbsent(body, b -> new CopyOnWriteArrayList<>()).add(System.nanoTime()); + if (body.startsWith("fail") && attempt <= failures) { + throw new IllegalStateException("Forced failure of " + body); + } + routed.add(body); + }); + } + }); + context.start(); + return context; + } + + private static void publish(String topic, String... bodies) throws Exception { + publish(topic, 1, false, bodies); + } + + /** + * Publishes the given bodies; a retained publish without bodies clears the retained message of the topic. + */ + private static void publish(String topic, int qos, boolean retained, String... bodies) throws Exception { + MqttClient publisher + = new MqttClient(brokerUrl(), "manual-ack-publisher-" + UUID.randomUUID(), new MemoryPersistence()); + try { + publisher.connect(); + if (retained && bodies.length == 0) { + publisher.publish(topic, new byte[0], qos, true); + } + for (String body : bodies) { + publisher.publish(topic, body.getBytes(StandardCharsets.UTF_8), qos, retained); + } + publisher.disconnect(); + } finally { + publisher.close(); + } + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 5b1e7e99e39d9..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 @@ -805,6 +805,22 @@ 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. For a QoS 1 message the +broker never sent it again during the connection, and after `receiveMaximum` such messages it sent the consumer +nothing more (the client completes the QoS 2 flow on its own, so QoS 2 messages did not stop the consumer). Now, for a +QoS 1 message on a session that the broker keeps after the disconnect (`cleanStart=false` and `sessionExpiryInterval` +greater than 0), the consumer connects again after a failed exchange (after one second, doubled for each failed +exchange in a row up to `maxReconnectDelay`, and back to one second after any exchange that succeeds; logged at WARN), +and the broker sends the message again. In all other +cases (a clean session, which is the default, a QoS 2 message, or `cleanStart=false` without `sessionExpiryInterval`, +where the session ends with the connection) the broker never sends the message again, so the failed message is +acknowledged and the failure is left to the Camel error handler. A consumer that uses a client set with the `client` +option leaves a failed QoS 1 message on a session that the broker keeps unacknowledged, as before; as the consumer does +not connect that client, set the `cleanStart` and `sessionExpiryInterval` of the endpoint to match its connection +options. + === 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..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 @@ -498,9 +498,14 @@ 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 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 69ea705e46257..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 @@ -666,9 +666,14 @@ 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 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. * @@ -687,9 +692,14 @@ 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 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. *