etzel-mes commented on PR #29421:
URL: https://github.com/apache/camel/pull/29421#issuecomment-6723644261
Thanks @huber-and, and thanks for writing up the drop behaviour so precisely
in the description. It made the review much easier, and the integration tests
already go further than camel-hivemq's, which has no connection-drop test at
all. I read the component against the vertx-mqtt 5.2.1 sources, ran the 67 unit
and 25 integration tests locally (Mosquitto 2.0.22 and HiveMQ CE, all green),
and then ran the component against EMQX 5.8.6 with real faults: a paused
broker, a broker restart and server-side disconnects, at QoS 1 and 2, MQTT
3.1.1 and 5, with a Paho client as the independent observer.
## Claus's points, seen live
- **In-flight slots (1).** With `ackTimeout=0` and four concurrent senders,
two connection losses were enough: every publish then failed with "Attempt to
exceed the limit of 10 inflight messages" while the connection was up and the
broker received nothing, until the application was restarted. With the default
`ackTimeout=30` the slots come back after 30 s.
- **Manual acks (2).** Consumer with `manualAcksEnabled`,
`receiveMaximum=20`, every third exchange failing: after 20 failures the
consumer stopped receiving, the broker held 20 in flight and queued the rest. A
reconnect delivered the 20 again, 20 more failed, and it stopped again. Nothing
was lost, but nothing moved either. Same picture at QoS 2.
- **Route on the event loop (3).** `maxPendingMessages=1500`, a burst of
1300 messages, 20 ms per exchange: from message 1010 on, the route ran on
`vert.x-eventloop-thread-0`.
- **Loss on stop (5).** Five route stop/start cycles at about 2000 msg/s,
QoS 1, auto ack, kept session: 3 messages were acknowledged and dropped
("without a subscriber"), and were missing at the end.
- **Stop during connect (6).** Route stopped while the first CONNECT was
waiting for its CONNACK: `NullPointerException` on the event loop, and the
broker showed the client as connected after the route had stopped.
## Two more things I found
1. **`publish()` can run off the connection's context.** `whenConnected()`
creates a context-less `Promise`; when the event loop completes it before the
caller has attached its `compose`, the compose runs on the caller's Camel
thread, and `MqttClient.publish()` is called from two threads at once. The
client assigns packet ids and topic aliases under its lock but writes outside
it, so the packets can leave in the wrong order. EMQX's trace showed it
directly: `PUBLISH(PacketId=2, Topic=, alias)` arrived before
`PUBLISH(PacketId=1, Topic=live/s1g, alias)`, EMQX answered with
`protocol_error` and closed the connection, 1.3 s after start. It happened in 2
of 3 starts with four concurrent senders, at QoS 1 as well as 2, MQTT 5 only
(3.1.1 has no aliases, but the same race touches the `unacknowledged` map,
which the class comment says needs no lock). Creating the promise on the
connection's context, or doing the whole publish inside `run(...)`, closes it.
2. **The producer completes the exchange on the event loop.**
`callback.done(false)` is called from the Vert.x thread, so the rest of the
route and the error handler run there, on the loop that also carries this
connection's socket and keep-alive. A 12 s blocking step after
`.to("vertx-mqtt:...")` with `keepAliveInterval=5` dropped every connection on
that loop, the producer's and an unrelated consumer's, on each cycle (9 losses
in 45 s). Camel's redelivery delay also slept on the loop (the blocked-thread
checker fired 70 times). camel-vertx-http completes the exchange on
`vertx.executeBlocking` for exactly this reason; the same here would do.
Smaller, seen live or in the client sources:
- After a reconnect, every exchange that waited in `whenConnected()` is
released at once; beyond `maxInflightQueue` they all fail immediately with the
in-flight limit error (about 140 per reconnect at 200 msg/s). Releasing them in
order against the limit, or documenting that the limit also bites after a
reconnect, would help.
- The client's keep-alive only watches writer idleness, so a producer that
publishes continuously never sends PINGREQ and does not notice a hung broker; a
12 s broker freeze went unnoticed, and only `ackTimeout` (QoS 1/2) would catch
it. That is vertx-mqtt's behaviour, not the component's, but worth a doc
sentence, and maybe an upstream issue next to #197.
- QoS 2 across a drop: the client does not resend with DUP, the exchange
fails, and Camel's redelivery sends a new packet id, so the broker can receive
the message twice. The description says redelivery is left to Camel, which is
fine, but the docs should say that QoS 2 is then at-least-once across a
reconnect.
- In vertx-mqtt 5.2.1, a PUBREL re-sent by the broker after a reconnect
finds no `qos2inbound` entry and gets no PUBCOMP, so with a kept session the
message is never completed and the broker re-sends the PUBREL on every
reconnect. Upstream, but it limits what manual acks can promise at QoS 2.
- `pause()` is buffered: a burst of 200 QoS 1 messages was read and
acknowledged in full before the pause took effect, so `maxPendingMessages` is a
soft bound and "stops reading" in the docs is a bit strong.
- vertx-mqtt's own docs mark the module and its MQTT 5 client as tech
preview; the component page could say so.
- A publish into a closed channel surfaces as a message-less
`StacklessClosedChannelException`.
## Where I would go slightly differently from the review
I tried three of the fixes on a scratch branch to see what they do; no code
here, just what I found.
- **Point 3.** Instead of a custom Camel thread pool profile, a Vert.x
worker of one thread (`createSharedWorkerExecutor(..., 1)` with ordered
`executeBlocking`) keeps the order, has no queue bound (the pause is the
bound), and nothing can fall back to the caller. One caution with the profile
route: `maxQueueSize <= 0` in Camel gives a `SynchronousQueue`, not an
unbounded queue. With the worker, the same 1300-message burst had zero
exchanges on the event loop.
- **Point 5.** Pause the client first, wait for the pending exchanges, then
unsubscribe and disconnect, and let the consumer implement `ShutdownAware` so
graceful shutdown sees the pending count and pauses in `prepareShutdown`. With
that order the same five stop/start cycles lost nothing over 74k messages.
- **Point 6.** Rather than guarding `client` for null, keep the client of
the attempt in a local reference in `connect()` and call `disconnect()`
unconditionally in `stop()`: the client answers at once when closed and closes
the socket itself when the CONNACK of a pending connect arrives. The
stop-during-connect case then ends with no exception and no client left at the
broker.
- **Point 2.** The camel-paho-mqtt5 comparison is a bit more nuanced: it
reconnects on a failed exchange only for QoS 1 with a kept session, and in
every other case it acknowledges the failed message so the broker's in-flight
slot is freed. Copying that second half is what prevents the stall above.
The rest of the review I agree with as written. Happy to run the next
revision against the same setup.
_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]
