[
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)
