[jira] [Resolved] (CAMEL-25793) camel-paho-mqtt5 - with manualAcksEnabled a failed message is never retried, and after receiveMaximum failures the consumer receives nothing more

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25793?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen resolved CAMEL-25793.
---------------------------------
    Resolution: Fixed

> camel-paho-mqtt5 - with manualAcksEnabled a failed message is never retried, and after receiveMaximum failures the consumer receives nothing more
> -------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-25793
>                 URL: https://issues.apache.org/jira/browse/CAMEL-25793
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-paho-mqtt5
>            Reporter: shashank
>            Assignee: shashank
>            Priority: Major
>             Fix For: 4.24.0
>
>
> {{PahoMqtt5Consumer}} (main 7aae27315946) with {{manualAcksEnabled=true}} adds an on-completion to every exchange ({{createExchange}}, lines 248-268): {{onComplete}} calls {{client.messageArrivedComplete(id, qos)}} (the PUBACK / PUBREC), {{onFailure}} only logs:
> {code:java}
> @Override
> public void onFailure(Exchange exchange) {
>     LOG.debug("Rollback due to error processing Exchange ID: {}", exchange.getExchangeId());
> }
> {code}
> The option documentation says the deferred acknowledgement "allows to rollback and retry the message in case of an error during routing in Camel". MQTT has no negative acknowledgement, though. The broker sends an unacknowledged PUBLISH again only when the client connects again to the same session (MQTT 5.0 section 4.4), and it never has more than the client's Receive Maximum unacknowledged QoS 1 / QoS 2 messages in flight to it (section 4.9; MQTT 3.1.1 brokers have their own in-flight window). So with the code on main:
> # the failed message is not retried while the connection lasts; with a clean session ({{cleanStart=true}}, the default) it is never sent again;
> # every failed message keeps one place of the receive window. After {{receiveMaximum}} failed exchanges (65535 by default; a broker may use a smaller in-flight window, and users lower it to limit the work in flight) the broker sends the consumer nothing more. The route stays started, nothing is logged above DEBUG, and the messages wait at the broker until the route is restarted.
> *Fix*: in {{onFailure}}: with {{cleanStart=false}} the consumer disconnects and connects again (on one scheduled executor per consumer, as the Paho client does not allow a disconnect from its callback) and subscribes again, so the broker sends the message again: the documented rollback and retry. The reconnects of failed exchanges in a row back off, so that a message that always fails does not make the consumer reconnect in a tight loop: the first waits the client's minimum automatic reconnect delay (1 s), each further one twice as long, up to {{maxReconnectDelay}}; a successful exchange resets it, and each such reconnect is logged at WARN. A reconnect that fails is scheduled again with the next delay. With a clean session, where the broker never sends the message again, the failed message is acknowledged, which frees its place; the failure is left to the Camel error handler as for any other exchange. A consumer that uses a client set with the {{client}} option is not reconnected (WARN, behaviour as before). New section "Manual acknowledgements" in the component page, upgrade note in the 4.24 guide.
> *Tests*: new {{PahoMqtt5ManualAcksFailureTest}} (4 tests, the real Paho client against an embedded Artemis MQTT broker from {{camel-test-infra-artemis}}, as in camel-paho; Artemis honours the client's Receive Maximum). {{failedMessagesDoNotStopTheConsumer}} ({{receiveMaximum=2}}; fail-1, fail-2 fail once, ok-3 and ok-4 follow) fails on main: {{Expecting actual: [] to contain exactly ["ok-3", "ok-4"]}} (nothing reaches the route after the two failures). {{failedMessageIsDeliveredAgainWithPersistentSession}} ({{cleanStart=false&sessionExpiryInterval=60}}) fails on main: {{Expecting actual: ["ok-2"] to contain exactly in any order ["fail-1", "ok-2"]}} (fail-1 is never retried). {{reconnectsAfterFailuresInARowBackOff}} (a message that fails three times, {{maxReconnectDelay=1500}}) checks with the fix that the attempts are at least 1 s, 1.5 s and 1.5 s apart and that the success resets the backoff; on main the message is never delivered again ({{Expecting actual: [] to contain exactly ["fail-3"]}}). {{acknowledgedMessagesKeepTheReceiveWindowOpen}} passes on main and with the fix: four successful messages through a window of 2, so the stall comes from the failures. camel-paho-mqtt5 unit tests: 21 run, 0 failures.
> *How it was found*: a TLA+ model of the manual acknowledgement protocol (broker in-flight window W, redelivery only on reconnect, Camel's onComplete / onFailure). The liveness property "every published message reaches the route" is violated with the code on main in 7 steps for N=4, W=2 (two failed messages fill the window, then stuttering); with the fix it holds (persistent session: every message is also routed successfully, no message dropped; N=6, W=3, 2 failures per message, 20,263 states). Then the JUnit test above.
> *Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x ({{onFailure}} only logs on all of them; read, not run). camel-paho (MQTT 3.1.1, deprecated) has the same {{onFailure}} and the same behaviour (brokers limit the unacknowledged messages in flight to an MQTT 3.1.1 client too, Artemis to 100 by default); it is not changed here.
> _Filed with Claude Code on behalf of allthingssecurity._



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)