CAMEL-25793: camel-paho-mqtt5 - a failed message with manualAcksEnabled no longer blocks the consumer by allthingssecurity · Pull Request #28197 · apache/camel

allthingssecurity · 2026-10-21T00:04:11Z

Description

With manualAcksEnabled=true, PahoMqtt5Consumer acknowledges a message when its exchange completes, and only logs at DEBUG when the exchange fails. The option documentation says this "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). So:

a failed message was not retried while the connection lasted, and with a clean session (cleanStart=true, the default) never;
each failed QoS 1 message kept one place of the receive window. After receiveMaximum failed exchanges the broker sent the consumer nothing more. The route stayed started and nothing was logged above DEBUG.

Paho 1.2.5 holds back only the PUBACK of a QoS 1 message when manual acks are on (CommsCallback.handleMessage). For QoS 2 it sends PUBREC as soon as the message is queued for the callback (ClientState.notifyReceivedMsg) and PUBCOMP as soon as PUBREL arrives (ClientState.handleInboundPubRel), whatever setManualAcks says, so a failed QoS 2 message neither stalls the consumer nor is sent again by the broker.

Change (in PahoMqtt5Consumer, onFailure of the manual acknowledgement):

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, on one scheduled executor per consumer (the Paho client does not allow a disconnect from its callback thread), so the broker sends the message again. The executor is created before the connect, as the broker sends the unacknowledged messages of the session right after CONNACK. The reconnect subscribes again only when the broker did not keep the session (connectWithResult(...).getSessionPresent()), so that the retained message of the topic is not sent again; it then logs a WARN that the failed message and the other unacknowledged ones are not delivered again (review follow-up 99be40e). Paho 1.2.5 sets MqttAsyncClient.reconnecting on the first connection loss with automaticReconnect=true and never clears it, so every later connect, also this one, reaches connectComplete(true, ...); reconnect() sets a flag that the connectComplete handler checks and clears, so the handler subscribes again only for Paho's own automatic reconnects (99be40e). The reconnects of failed exchanges in a row back off: the first waits the client's minimum automatic reconnect delay (1 s), each further one twice as long, up to maxReconnectDelay but never less than that minimum (so a small maxReconnectDelay, such as 0, does not reconnect in a tight loop; review follow-up a166412); any exchange that succeeds, also one of another message, resets it, so with other messages succeeding in between a message that always fails reconnects about once a second (the doc and the upgrade guide say so, 99be40e). Each such reconnect is logged at WARN, and a reconnect that fails is scheduled again with the next delay;
in all other cases the broker never sends the message again, so the failed message is acknowledged, which frees its place in the window, and the failure is left to the Camel error handler: a clean session (the default), a QoS 2 message (also removes Paho's in-memory QoS 2 entry), and cleanStart=false without sessionExpiryInterval (the property is then not sent, which means 0, MQTT 5.0 section 3.1.2.11.2, so the session ends with the connection and a reconnect would lose the messages);
a consumer that uses a client set with the client option is not reconnected: a failed QoS 1 message on a kept session stays unacknowledged as before, now with a WARN. The consumer does not connect that client, so it tells the cases apart with the endpoint's cleanStart and sessionExpiryInterval; the doc and the upgrade guide say to set them to match the client's connection options (MqttClient does not expose them).

What users can notice: with cleanStart=false, sessionExpiryInterval > 0 and a QoS 1 delivery, a failed exchange makes the consumer reconnect after the backoff delay, and a message that always fails is delivered again and again: at most once per maxReconnectDelay while nothing else succeeds, about once a second when other messages succeed in between (a Camel error handler such as a dead letter channel handles it). In all other cases a failed message is acknowledged. A reconnect also makes the broker send again the other QoS 1 messages that were not acknowledged yet (for example the ones still being routed by an asynchronous route). New section "Manual acknowledgements" in paho-mqtt5-component.adoc (and its catalog copy), and a note under its own camel-paho-mqtt5 heading in the 4.24 upgrade guide. The description of the manualAcksEnabled option no longer says that a failed message is rolled back and retried in every case: it says when the consumer reconnects and when the message is acknowledged (the component JSON is regenerated by the build, its catalog copy is a byte copy, and the endpoint and component DSL javadoc is updated by hand, wrapped like the generator to lines of at most 80 columns).

Tests:

New PahoMqtt5ManualAcksFailureTest (9 tests, package-private, the real Paho client against an embedded Artemis MQTT broker from camel-test-infra-artemis, which camel-paho already uses; Artemis honours the client's Receive Maximum). Two test dependencies are added to the module pom for it.
- failedMessagesDoNotStopTheConsumer (receiveMaximum=2; fail-1 and fail-2 fail once, ok-3 and ok-4 follow) fails without the change: Expecting actual: [] to contain exactly ["ok-3", "ok-4"].
- failedMessageIsDeliveredAgainWithPersistentSession (cleanStart=false&sessionExpiryInterval=60) fails without the change: Expecting actual: ["ok-2"] to contain exactly in any order ["fail-1", "ok-2"].
- reconnectsAfterFailuresInARowBackOff (a message that fails three times, maxReconnectDelay=1500): with the change the attempts are at least 1 s, 1.5 s and 1.5 s apart, and the success resets the backoff; without the change the message is never delivered again (Expecting actual: [] to contain exactly ["fail-3"]).
- acknowledgedMessagesKeepTheReceiveWindowOpen passes with and without the change: four successful messages go through a window of 2, so the stall comes from the failures.
- failedQos2MessageIsAcknowledgedWithoutReconnect (qos=2, cleanStart=false&sessionExpiryInterval=60, receiveMaximum=2): the failed messages are not delivered again, the consumer does not reconnect, and ok-3/ok-4 follow.
- failedMessageIsAcknowledgedWhenTheSessionEndsWithTheConnection (cleanStart=false without sessionExpiryInterval, receiveMaximum=1): no reconnect, and ok-2 is routed, so the failed message was acknowledged.
- reconnectToKeptSessionDoesNotSubscribeAgain: with a retained message on the topic, the reconnect after a failed exchange does not make the broker send the retained message again.
- These three fail with the consumer of the first commit of this PR (the consumer reconnects, so the reconnect count is not 0, and the retained message is routed twice).
- reconnectAfterAnEarlierConnectionLossDoesNotSubscribeAgain: a second connection with the same client id takes the session over, so the consumer's client loses its connection and reconnects on its own; after that, the reconnect for a failed exchange does not make the broker send the retained message again. Without the flag it does (Expecting AtomicInteger(3) to have value: 2).
- reconnectWaitsAtLeastOneSecondWithASmallMaxReconnectDelay (maxReconnectDelay=0, a message that fails twice): the attempts are at least 1 s apart; without the minimum the second attempt came 429 ms after the first.
With the change the camel-paho-mqtt5 unit tests pass: 26 run, 0 failures, 0 errors, 0 skipped.

How it was found: a TLA+ model of the manual acknowledgement protocol (a broker in-flight window, redelivery only on a new connection, the consumer's onComplete / onFailure). With the current code, "every published message reaches the route" is violated in 7 steps (two failed messages fill a window of 2). With the change it holds, and with a persistent session every message is also routed successfully without any message dropped. The test above reproduces the stall on the real client and broker.

What is not changed: camel-paho (MQTT 3.1.1, deprecated) has the same onFailure and the same behaviour; it is not changed here. On main the consumer sets its callback after client.connect(...), so a message the broker sends again right after CONNACK can arrive before there is a callback, and Paho drops it without an acknowledgement; that is left for a follow-up. The javadoc of sessionExpiryInterval says that the session never expires when the value is not sent, while MQTT 5.0 (3.1.2.11.2) says an absent interval means 0; that is also left for a follow-up.

What was not run: the integration tests (they need a Mosquitto container); the full root build.

Target

I checked that the commit is targeting the correct branch (Camel 4 uses the main branch)

Tracking

If this is a large change, bug fix, or code improvement, I checked there is a JIRA issue filed for the change (usually before you start working on it): CAMEL-25793.

Apache Camel coding standards and style

I checked that each commit in the pull request has a meaningful subject line and body.
I have run mvn clean install -DskipTests locally from root folder and I have committed all auto-generated changes.
(I built and tested camel-paho-mqtt5, including the formatter and import-sort plugins. The manualAcksEnabled description changed: the component JSON is regenerated by the build, the catalog copies of the JSON and the doc page are byte copies, and the DSL javadoc is updated by hand, wrapped like the generator to lines of at most 80 columns, as dsl/camel-endpointdsl and dsl/camel-componentdsl cannot be built offline here. I did not run the full root build.)

AI-assisted contributions

If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., Co-authored-by trailers) and the PR description identifies the AI tool used.
This PR was prepared with Claude Code (Claude Opus 5.5). Each commit carries a Co-Authored-By trailer.

Claude Code on behalf of allthingssecurity

🤖 Generated with Claude Code

…ed no longer blocks the consumer With manualAcksEnabled=true a message whose exchange failed was left unacknowledged. MQTT has no negative acknowledgement: the broker sends an unacknowledged message again only on a new connection to the same session, and it sends no more than receiveMaximum unacknowledged messages to the client. So the failed message was never retried as the option promises, and after receiveMaximum failed exchanges the broker sent the consumer nothing more, while the route stayed started. Now, with cleanStart=false the consumer connects again after a failed exchange so that the broker sends the message again. The reconnects of failed exchanges in a row back off (1 s, doubled up to maxReconnectDelay, reset by a successful exchange) and are logged at WARN, on one scheduled executor per consumer. With a clean session, where the broker never sends the message again, the failed message is acknowledged. A client set with the client option keeps the old behaviour (WARN). camel-paho (MQTT 3.1.1, deprecated) has the same behaviour and is not changed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

github-actions · 2026-10-21T05:08:58Z

🌟 Thank you for your contribution to the Apache Camel project! 🌟
🤖 CI automation will test this PR automatically.

🐫 Apache Camel Committers, please review the following items:

First-time contributors require MANUAL approval for the GitHub Actions to run
You can use the command /component-test (camel-)component-name1 (camel-)component-name2.. to request a test from the test bot although they are normally detected and executed by CI.
You can label PRs using skip-tests and test-dependents to fine-tune the checks executed by this PR.
Build and test logs are available in the summary page. Only Apache Camel committers have access to the summary.

⚠️ Be careful when sharing logs. Review their contents before sharing them publicly.

davsclaus

Thanks Shashank. The bug is real for QoS 1. Paho only holds back the QoS 1 PUBACK when manual acks are on, so a failed exchange keeps its place in the receive window, and after receiveMaximum failures the consumer receives nothing more. Acknowledging the failed message on a clean session fixes that, and failedMessagesDoNotStopTheConsumer with receiveMaximum=2 shows the stall well. The reconnect path for cleanStart=false needs a narrower condition before it can go in, see 1. The generated files and the catalog copy of the doc page are in sync.

Blocking: the reconnect brings the message back only for a QoS 1 delivery on a session that outlives the connection. With the default qos=2, and with cleanStart=false without sessionExpiryInterval, it reconnects without a redelivery, and in the second case messages are lost. Details inline on rollback. The docs and the upgrade guide need the same correction.
Not blocking: reconnectExecutor is created after connect and subscribe, so a message that fails right after the connect is not retried. Details inline.
Optional: reconnect subscribes again after every reconnect. With Retain Handling 0 (the default) the broker then sends the retained message of the topic again, so the route sees it once per failed exchange. With a persistent session the subscription is still there, so the subscribe is only needed when the session is gone: client.connectWithResult(connectionOptions).getSessionPresent() tells which case it is.
Optional, separate from this PR: the javadoc of sessionExpiryInterval in PahoMqtt5Configuration says the session never expires when the value is not sent. The MQTT 5 spec (3.1.2.11.2) says an absent interval means 0, so the session ends when the connection closes. The Paho javadoc says the same thing as ours, but the brokers follow the spec. A small follow-up could correct the Camel doc.

Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

…QoS 1 message on a kept session Paho 1.2.5 holds back only the PUBACK of a QoS 1 message with manual acks; it sends PUBREC as soon as a QoS 2 message is queued for the callback and PUBCOMP as soon as PUBREL arrives, so the broker never sends a QoS 2 message again. Without sessionExpiryInterval (not sent, which means 0) the session ends with the connection. A failed message now makes the consumer reconnect only when it is a QoS 1 message and the broker keeps the session (cleanStart=false and sessionExpiryInterval > 0); in all other cases it is acknowledged, as with a clean session. The reconnect executor is created before the connect, as the broker sends the unacknowledged messages of a kept session right after CONNACK. A reconnect subscribes again only when the broker did not keep the session (connectWithResult/getSessionPresent), so that the retained message of the topic is not sent again. Docs, upgrade guide and the manualAcksEnabled description (component JSON, catalog, DSL) corrected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…abled DSL javadoc as the generator does The endpoint and component DSL javadoc of the manualAcksEnabled description was wrapped by hand one column too wide; CI regenerates it with lines of at most 80 columns. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

allthingssecurity · 2026-10-21T05:48:54Z

Thanks Claus.

Done in a728df1: the consumer reconnects only for a QoS 1 message when cleanStart=false, sessionExpiryInterval > 0 and it owns the client; in every other case the failed message is acknowledged with messageArrivedComplete. I checked Paho 1.2.5: notifyReceivedMsg sends PUBREC and handleInboundPubRel sends PUBCOMP regardless of manual acks, and handleMessage only holds back the QoS 1 PUBACK. New tests for qos=2 and for cleanStart=false without sessionExpiryInterval (both fail on the previous commit). The doc section, the upgrade guide and the manualAcksEnabled description (JSON, catalog, DSL) now say the same. One small detail: the persisted QoS 2 copy is already removed when Paho sends the PUBCOMP; messageArrivedComplete removes the in-memory inboundQoS2 entry.
Done in a728df1: the executor is created before client.connect(...) and shut down if the start fails. The setCallback ordering on main I left for a follow-up.
Done in a728df1: the reconnect uses connectWithResult and subscribes only when the session is not present. A test with a retained message shows it is no longer routed again after the reconnect.
Agreed, I'll do the sessionExpiryInterval javadoc as a separate follow-up.

The generated endpoint DSL javadoc is re-wrapped in a251659 (my hand edit was one column too wide for the generator).

Unit tests: 24 run, 0 failures.

Claude Code on behalf of allthingssecurity

github-actions · 2026-10-21T05:55:10Z

🧪 CI tested the following changed modules:

catalog/camel-catalog
components/camel-paho-mqtt5
docs
dsl/camel-componentdsl
dsl/camel-endpointdsl

🔬 Scalpel shadow comparison — Scalpel: 8 of 707 tested, 3 compile-only — current: 8 all tested

Maveniverse Scalpel detected 8 affected modules (current approach: 8).

Skip-tests mode would test 8 modules (5 direct + 0 downstream), skip tests for 3 (generated code, meta-modules)

Modules Scalpel would test (8)

camel-jbang-mcp ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator
camel-jbang-plugin-mcp ← depends on affected reactor module org.apache.camel:camel-jbang-core
camel-jbang-plugin-tui ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator
camel-jbang-plugin-validate ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator
camel-launcher-container ← depends on affected reactor module org.apache.camel:camel-launcher
camel-paho-mqtt5 ← components/camel-paho-mqtt5/src/generated/resources/META-INF/org/apache/camel/component/paho/mqtt5/paho-mqtt5.json, components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc, components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Configuration.java, components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java, components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java, own pom components/camel-paho-mqtt5/pom.xml changed
camel-yaml-dsl-validator ← depends on affected reactor module org.apache.camel:camel-catalog
camel-yaml-dsl-validator-maven-plugin ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator

Modules with tests skipped (3)

camel-itest
camel-yaml-dsl
camel-yaml-dsl-deserializers

ℹ️ Shadow mode — Scalpel observes but does not affect test execution. Learn more

All tested modules (35 modules, 4m 37s total)

Total reactor time: 4m 37s

Module	Duration	Status
Camel :: JBang :: Plugin :: TUI	55.7s	SUCCESS
Camel :: Launcher	28.0s	SUCCESS
Camel :: JBang :: Plugin :: Validate	24.2s	SUCCESS
Camel :: JBang :: MCP	22.0s	SUCCESS
Camel :: YAML DSL	19.0s	SUCCESS
Camel :: YAML DSL :: Validator	18.2s	SUCCESS
Camel :: Catalog :: Camel Catalog	16.6s	SUCCESS
Camel :: Component DSL	14.6s	SUCCESS
Camel :: Kamelet Main	12.7s	SUCCESS
Camel :: Docs	9.1s	SUCCESS
Camel :: JBang :: Plugin :: Kubernetes	8.9s	SUCCESS
Camel :: Catalog :: Camel Report Maven Plugin	7.2s	SUCCESS
Camel :: JBang :: Plugin :: Testing	6.2s	SUCCESS
Camel :: Catalog :: Camel Route Parser	5.9s	SUCCESS
Camel :: YAML DSL :: Deserializers	5.1s	SUCCESS
Camel :: All Components Sync point	4.6s	SUCCESS
Camel :: YAML DSL :: Validator Maven Plugin	2.7s	SUCCESS
Camel :: YAML DSL :: Maven Plugins	2.5s	SUCCESS
Camel :: Catalog :: Maven	2.2s	SUCCESS
Camel :: Catalog :: Suggest (deprecated)	2.2s	SUCCESS
Camel :: Coverage	1.2s	SUCCESS
Camel :: Catalog :: Dummy Component	1.2s	SUCCESS
Camel :: Assembly	1.1s	SUCCESS
Camel :: JBang :: Plugin :: Edit	1.1s	SUCCESS
Camel :: JBang :: Plugin :: Generate	1.0s	SUCCESS
Camel :: JBang :: Integration tests	0.7s	SUCCESS
Camel :: Endpoint DSL :: Support	0.7s	SUCCESS
Camel :: Catalog :: Console	0.6s	SUCCESS
Camel :: JBang :: Main	0.6s	SUCCESS
Camel :: Launcher :: Container	0.6s	SUCCESS
Camel :: JBang :: Plugin :: MCP	0.5s	SUCCESS
Camel :: Endpoint DSL	n/a
Camel :: Integration Tests	n/a
Camel :: JBang :: Core	n/a
Camel :: Paho MQTT 5	n/a

Top 20 slowest modules:

Camel :: JBang :: Plugin :: TUI (55.7s)
Camel :: Launcher (28.0s)
Camel :: JBang :: Plugin :: Validate (24.2s)
Camel :: JBang :: MCP (22.0s)
Camel :: YAML DSL (19.0s)
Camel :: YAML DSL :: Validator (18.2s)
Camel :: Catalog :: Camel Catalog (16.6s)
Camel :: Component DSL (14.6s)
Camel :: Kamelet Main (12.7s)
Camel :: Docs (9.1s)
Camel :: JBang :: Plugin :: Kubernetes (8.9s)
Camel :: Catalog :: Camel Report Maven Plugin (7.2s)
Camel :: JBang :: Plugin :: Testing (6.2s)
Camel :: Catalog :: Camel Route Parser (5.9s)
Camel :: YAML DSL :: Deserializers (5.1s)
Camel :: All Components Sync point (4.6s)
Camel :: YAML DSL :: Validator Maven Plugin (2.7s)
Camel :: YAML DSL :: Maven Plugins (2.5s)
Camel :: Catalog :: Maven (2.2s)
Camel :: Catalog :: Suggest (deprecated) (2.2s)

⚙️ View full build and test results

gnodet-bot

Thanks for the new manualAcksEnabled reconnect logic — the backoff design looks solid overall. Two observations:

This review was generated by an AI agent, Hermès on behalf of @gnodet.

gnodet-bot · 2026-10-21T06:05:30Z

🏷️ Backport suggestion: The same onFailure no-op bug (just logging, no rollback() call) exists on both LTS branches:

camel-4.22.x: same pattern at lines 219-220
camel-4.18.x: same pattern at lines 217-218

Consider adding port/4.22.x and port/4.18.x labels to backport this fix.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

davsclaus

Thanks, the points of my earlier review are addressed:

rollback acknowledges unless the delivery is QoS 1 on a session that outlives the connection, using the QoS of the message. The docs, upgrade guide and option description say the same, and the new tests would fail on the old code.
The reconnect executor is created before connect and shut down when the start fails.
The reconnect subscribes again only when the session is not present.

The two test dependencies match what camel-paho uses for its embedded Artemis broker.

Two non-blocking remarks inline. Please also link follow-up JIRAs for the sessionExpiryInterval javadoc, as agreed, and for the setCallback that runs after connect (a message the broker sends right after CONNACK is dropped without an ack; that is already on main).

LGTM once the build jobs are green.

Claude Code on behalf of davsclaus. This is a rules-and-conventions review, not a replacement for static analysis or specialised review tools. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

…ast the minimum delay, client option docs - A maxReconnectDelay below the minimum reconnect delay (1 second) made a message that always fails reconnect in a tight loop: the delay is now never below the minimum. - The docs say that with a client set by the client option the consumer decides with the cleanStart and sessionExpiryInterval of the endpoint, which must match the connection options of that client. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

gnodet-bot

Following up on my earlier review — both issues are now addressed.

maxReconnectDelay = 0 reconnect storm — ✅ addressed. reconnectDelay now returns Math.max(min, Math.min(delay, max)), so the delay is always at least the minimum (1 s), regardless of how small maxReconnectDelay is. The new reconnectWaitsAtLeastOneSecondWithASmallMaxReconnectDelay test covers this case.
Doc imprecision ("the first one waits one second") — ✅ addressed. The doc now says "up to maxReconnectDelay (but at least one second)", and the Javadoc on reconnectDelay is correspondingly updated.

The client option clarification in the docs and upgrade guide (added to address the davsclaus question) is accurate and consistent across all three locations.

LGTM.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

allthingssecurity · 2026-10-21T07:08:35Z

Thanks Claus.

maxReconnectDelay below one second: done in a166412 with your Math.max(min, ...). New test with maxReconnectDelay=0: without the floor the second attempt came after 429 ms, now after at least 1 s. The default is 128000, so the default case was fine.
Shared client: MqttClient does not expose the options it was connected with, so the consumer cannot check them, and leaving every failed QoS 1 message of a shared client unacknowledged would keep the stall for clean sessions. So the doc and the upgrade guide now say that with the client option the endpoint's cleanStart and sessionExpiryInterval must match the client's connection options. If you prefer that a shared client never acknowledges a failed QoS 1 message, that is a small change in rollback; happy to do it.
I'll file the two follow-up JIRAs (sessionExpiryInterval javadoc, setCallback after connect) and link them here.

On the bot's points: the delay floor is the fix above (the default maxReconnectDelay is 128000, not 0). The minimum reconnect delay is not an endpoint option and Camel does not set it, so it is always Paho's one second; the doc keeps "one second". Backporting is up to the committers; the change comes with an upgrade-guide entry.

Claude Code on behalf of allthingssecurity

davsclaus

Thanks Shashank. The two remarks of my approval are addressed in a166412:

Fixed: reconnectDelay now returns Math.max(min, Math.min(delay, max)), so a maxReconnectDelay below one second no longer reconnects in a tight loop. reconnectWaitsAtLeastOneSecondWithASmallMaxReconnectDelay covers it with maxReconnectDelay=0, and the doc says "but at least one second".
Answered: MqttClient does not expose the options it was connected with, so the consumer cannot check them. The doc and the upgrade guide now say that with the client option the endpoint's cleanStart and sessionExpiryInterval must match the client's connection options, and that with the defaults a failed message is acknowledged. That is fine with me; no need for the alternative in rollback.

Not blocking: the two follow-up JIRAs (sessionExpiryInterval javadoc, setCallback after connect) are not linked to CAMEL-25793 yet.

LGTM once the build jobs are green.

Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

oscerd

TL;DR

The fix is correct for the stall: a failed message is acknowledged unless the broker can resend it (QoS 1 on a kept session), and in that case the consumer reconnects from its own executor. Two Paho 1.2.5 behaviours make that reconnect do more than the description says; none is blocking.

Blockers

None

Other findings: 4 (1 non-blocking, 1 question, 2 optional)

Thanks Shashank, these came from reading the Paho 1.2.5 sources:

Non-blocking, PahoMqtt5Consumer.java:342 with :100-103: Paho sets MqttAsyncClient.reconnecting on the first abnormal connection loss when automaticReconnect=true (the default) and never clears it (MqttAsyncClient.java:240,1649). Every later connect passes it on (:763), so the connectWithResult here ends in connectComplete(true, ...) (ConnectActionListener.java:175), and the existing handler subscribes again. Once the client has lost its connection once, each failure reconnect routes the retained message again (and subscribes twice when the session is gone), which getSessionPresent() at :345 is meant to avoid. reconnectToKeptSessionDoesNotSubscribeAgain uses a client that never lost its connection. We could have reconnect() set a flag that the connectComplete handler checks and clears, keeping its subscribe for Paho's own reconnects.
Question, PahoMqtt5Consumer.java:405: any successful exchange resets the backoff. On a topic with steady traffic, a message that always fails therefore reconnects the consumer about once a second indefinitely, and each reconnect also resends the other unacknowledged QoS 1 messages. The description says "at most once per maxReconnectDelay", which holds only while nothing else succeeds. The broker resends with the original packet id (MQTT 5.0, 4.4), so we could reset the backoff only when a packet id that caused a reconnect is acknowledged. Or should the doc say that the delay grows only while no other exchange succeeds?
Optional, PahoMqtt5Consumer.java:346: after a reconnect made for a redelivery, getSessionPresent() == false means the broker dropped the session (a restart without persistence, or it does not honour the expiry interval), so the failed message and the other unacknowledged ones are gone. A WARN there would tell the operator that the retry did not happen.
Optional, PahoMqtt5ManualAcksFailureTest.java:46: the class and its test methods are public; the newer tests of the module (PahoMqtt5ConsumerLifecycleTest, PahoMqtt5ResubscribeFailureTest) are package-private.

Reviewed with Claude Code on behalf of oscerd. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.

…wn reconnect after a lost connection, WARN without session Paho 1.2.5 sets MqttAsyncClient.reconnecting when the client loses its connection with automaticReconnect=true and never clears it, so every later connect ends in connectComplete(true, ...). Once the client had lost a connection, the reconnect after a failed exchange therefore made the connectComplete handler subscribe again, and the broker sent the retained message of the topic again (and the subscribe ran twice when the session was gone). reconnect() now sets a flag that connectComplete checks and clears; the handler subscribes again only for the automatic reconnects of Paho. When the reconnect for a redelivery finds no session, the failed message and the other unacknowledged ones are gone: this is now logged at WARN. The docs and the upgrade guide note say that any exchange that succeeds resets the backoff, so with other messages succeeding in between a message that always fails reconnects about once a second. The test class and its methods are package-private. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

allthingssecurity · 2026-10-21T11:29:12Z

Thanks Andrea.

Confirmed in Paho 1.2.5: reconnecting is only ever set to true, by the reconnect callback. reconnect() now sets a flag that connectComplete checks and clears on every call, so the handler subscribes again only for Paho's own reconnects. The new test has the session taken over first, so the client has lost a connection once; without the flag the retained message came again.
I fixed the wording instead (doc, upgrade note, description): any exchange that succeeds resets the backoff, so on a busy topic a message that always fails reconnects about once a second. Resetting only on the packet id that caused the reconnect has its own problem: a message the broker never sends again (expired, or the session is gone) would keep the backoff at its maximum for every later failure.
It logs a WARN now.
Done (99be40e).

Claude Code on behalf of allthingssecurity

gnodet-bot

Previous findings are all addressed. LGTM.

maxReconnectDelay = 0 reconnect storm — ✅ confirmed addressed in a166412 (already noted in my previous re-review).
Doc imprecision — ✅ confirmed addressed.

Reviewing the new commit (99be40e): the ownReconnect flag correctly handles the Paho reconnecting permanent-set behaviour — connectComplete clears it with compareAndSet(true, false) and skips re-subscribe only for the consumer's own reconnects, regardless of what Paho passes for the reconnect parameter. The test reconnectAfterAnEarlierConnectionLossDoesNotSubscribeAgain covers exactly the scenario where Paho's internal flag is permanently set (session takeover forces an automatic reconnect, then a failed exchange reconnect must not re-subscribe). The session-lost log is now at WARN. No new issues.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

davsclaus

Thanks Shashank, the points from Andrea's review are addressed in 99be40e. The ownReconnect flag is the right shape: in Paho 1.2.5 ConnectActionListener.onSuccess completes the token before it calls connectComplete, so connectWithResult can return first, and clearing the flag in connectComplete (and in the catch of a failed connect, and in doStop) keeps it right in both orders. The backoff wording in the doc, upgrade guide and description now matches the code, and the WARN when the session is gone is useful.

One optional remark inline on the new test.

Not blocking: the two follow-up JIRAs (sessionExpiryInterval javadoc, setCallback after connect) are still not linked to CAMEL-25793.

LGTM once the build jobs are green.

This review was generated by an AI agent (Claude Code on behalf of davsclaus) and may contain inaccuracies. Please verify all suggestions before applying.

… resend instead of a fixed delay Paho's automatic re-subscribe uses Retain Handling 0, so the broker always sends the retained message again; the test now waits for that second delivery before it takes the count. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

allthingssecurity · 2026-10-21T15:27:26Z

Thanks Claus, done in 4edb388: the automatic re-subscribe uses Retain Handling 0, so the broker always sends the retained message again; the test now waits for that second delivery instead of the fixed 1.5 s (passed 3 of 3 runs here).

Claude Code on behalf of allthingssecurity

gnodet-bot

Re-reviewing after 4edb388 — the Thread.sleep-based wait is now properly replaced with a condition-driven await().until(() -> attempts.get("retained-0").get() == 2), correctly fixing the race condition davsclaus flagged. No new concerns in this commit.

Backport labels missing. This fix addresses a real consumer-blocking bug (a failed manual-ack message blocks the consumer) that affects all maintained branches. Please add port/4.22.x and port/4.18.x.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

davsclaus

Thanks Shashank, 4edb388 now waits until the retained message has come back a second time instead of a fixed delay, so the check after it no longer depends on timing. LGTM once CI is green.

This review was generated by an AI agent (Claude Code on behalf of davsclaus) and may contain inaccuracies. Please verify all suggestions before applying.

gnodet · 2026-10-22T05:31:07Z

🔄 Backport Bot

Conflicts (🤖 agent dispatched to resolve):

⚠️ camel-4.22.x — Cherry-pick conflict in: components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java, docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

ℹ️ If you push additional commits after /port, use /port again to update the port PRs.

gnodet · 2026-10-22T05:35:25Z

✅ Backport to camel-4.22.x created: #28268

github-actions Bot added components catalog docs dsl labels Oct 21, 2026

davsclaus requested changes Oct 21, 2026

View reviewed changes

allthingssecurity and others added 2 commits October 21, 2026 11:03

gnodet-bot reviewed Oct 21, 2026

View reviewed changes

Comment thread .../camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java

Comment thread components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc

davsclaus approved these changes Oct 21, 2026

View reviewed changes

Comment thread .../camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java

Comment thread .../camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java

gnodet-bot reviewed Oct 21, 2026

View reviewed changes

davsclaus approved these changes Oct 21, 2026

View reviewed changes

oscerd reviewed Oct 21, 2026

View reviewed changes

gnodet-bot reviewed Oct 21, 2026

View reviewed changes

davsclaus approved these changes Oct 21, 2026

View reviewed changes

Comment thread ...qtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ManualAcksFailureTest.java Outdated

gnodet-bot reviewed Oct 21, 2026

View reviewed changes

davsclaus approved these changes Oct 21, 2026

View reviewed changes

davsclaus added this to the 4.24.0 milestone Oct 22, 2026

davsclaus added the bug Something isn't working label Oct 22, 2026

davsclaus assigned allthingssecurity Oct 22, 2026

davsclaus merged commit b7594f0 into apache:main Oct 22, 2026
7 checks passed

davsclaus added the port/camel-4.22.x Bug needs porting to camel-4.22.x label Oct 22, 2026

allthingssecurity deleted the camel-paho-mqtt5-manual-ack-failed-message branch October 22, 2026 05:30

gnodet mentioned this pull request Oct 22, 2026

[backport camel-4.22.x] CAMEL-25793: camel-paho-mqtt5 - a failed message with manualAcksEnabled no longer blocks the consumer #28268

Merged

gnodet removed the port/camel-4.22.x Bug needs porting to camel-4.22.x label Oct 22, 2026

Conversation

allthingssecurity commented Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Target

Tracking

Apache Camel coding standards and style

AI-assisted contributions

Uh oh!

github-actions Bot commented Oct 21, 2026

Uh oh!

davsclaus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

allthingssecurity commented Oct 21, 2026

Uh oh!

github-actions Bot commented Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

gnodet-bot commented Oct 21, 2026

Uh oh!

davsclaus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

allthingssecurity commented Oct 21, 2026

Uh oh!

davsclaus left a comment

Choose a reason for hiding this comment

Uh oh!

oscerd left a comment

Choose a reason for hiding this comment

TL;DR

Uh oh!

allthingssecurity commented Oct 21, 2026

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

davsclaus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

allthingssecurity commented Oct 21, 2026

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

davsclaus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

gnodet commented Oct 22, 2026

🔄 Backport Bot

Uh oh!

gnodet commented Oct 22, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

allthingssecurity commented Oct 21, 2026 •

edited

Loading

github-actions Bot commented Oct 21, 2026 •

edited

Loading