etzel-mes commented on PR #29421:
URL: https://github.com/apache/camel/pull/29421#issuecomment-6724653228
Thanks @huber-and, that was quick. I ran 010b8b2f49f against the same EMQX
5.8.6 setup as before (paused broker, server-side kicks, route stop/start,
concurrent senders, MQTT 5 and QoS 1/2), after the 79 unit and 25 integration
tests passed locally.
## What the live run shows now
- **Publish race:** three starts with four concurrent QoS 2 senders at 200
msg/s, 25 s each: no `protocol_error`, no disconnect, 0 failed, 0 duplicates.
Before, 2 of 3 starts dropped within 1.3 s.
- **Blocking step after the producer (12 s, `keepAliveInterval=5`):** the
rest of the route ran on `vert.x-worker-thread-*`, 0 connection losses in 45 s
(before: 9), no blocked-thread warnings.
- **In-flight slots, `ackTimeout=0`, three kicks:** failures only around
each kick, none in the steady state afterwards, broker shows `inflight_cnt=0`
every time. Before, the limit error was permanent after the second loss.
- **Route on the event loop (`maxPendingMessages=1500`, burst 1300, 20
ms):** 0 hits, everything on the consumer's worker thread (before: from message
1010 on).
- **Stop/start, five cycles at ~2000 msg/s, QoS 1, kept session:** 0
acknowledged-and-dropped, 0 missing over 78,957 messages, no drain timeout
(before: 3 lost).
- **Stop during CONNECT:** no exception, no client left at the broker.
- **Manual acks, QoS 2, every third exchange failing, `receiveMaximum=20`:**
no stall any more, 529 messages in 30 s, failed ones acknowledged (before:
stuck at 20).
## Manual acks at QoS 1 with a kept session: the stall comes back in another
shape
Same scenario as before (QoS 1, `cleanStart=false`,
`sessionExpiryInterval=300`, `receiveMaximum=20`, every third exchange
failing). The first failure after each connect closes the connection, EMQX
redelivers the 20 in flight about a second later, and the window fills up with
messages that fail and messages whose ack came too late: from the tenth second
on, every reconnect delivered the same 20 `dup=true` messages and nothing new,
once a second, for the rest of the run. 26 connects in 30 s, 40 distinct
messages seen, 466 duplicates, `mqueue_len` at the broker 116 → 307 → 497.
Closing the connection for redelivery is the right default, and a message that
always fails is the route's problem: with
`onException(Exception.class).handled(true)` in the route the same run
processed 535 messages, 0 duplicates, 1 connect. But I found one thing I would
fix before merge, and three I would do at some point.
**Before merge, since it loses messages with the defaults:**
`canRedeliver()` only checks `cleanStart=false`. With MQTT 5, the default, and
`sessionExpiryInterval` at its default of 0, the session ends with the
connection, so the close loses the message instead of having it delivered
again. The redelivery path needs `sessionExpiryInterval > 0` as well (or MQTT
3.1.1); otherwise acknowledging is the better of the two. And a sentence in the
docs next to "a route must tolerate duplicates": a message whose exchange fails
every time comes back after every reconnect, and the route's error handler, a
dead letter channel for example, has to complete it.
**Worth doing, not blocking:**
- The disconnect is immediate, and `acknowledge()` then refuses the acks of
the exchanges that complete after it ("Not acknowledging the message … of an
earlier connection", 64 times in the run), so messages that succeeded are
redelivered as well; 9 of the 20 in the stuck window were such. Letting the
exchanges already in flight finish and acknowledge before the close keeps the
window to the messages that actually failed.
- A redelivering close is logged at DEBUG only and is not reported to the
exception handler or the health check, so the only visible trace of the loop is
"Connected to the MQTT broker" once a second. A WARN, or the usual report,
would make it findable.
- The reconnect delay is reset on every successful connect, so a message
that keeps failing reconnects once a second forever. Backing off while the
failures repeat would help.
Also seen, as now documented: after a kick, the exchanges that waited for
the connection fail beyond `maxInflightQueue` (about 180 per kick at 200
msg/s). Fine with the doc sentence.
Apart from this, nothing from my side: the component behaved under every
other fault I could inject, and with the expiry check in I am happy with it.
Thanks for taking the suggestions on so thoroughly.
_Claude Code on behalf of etzel-mes. This review was generated with an AI
agent and checked by me; it may still contain inaccuracies, please verify
before applying._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
