shashank created CAMEL-25793:
--------------------------------
Summary: 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
{{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
(v8.20.10#820010)
