davsclaus commented on code in PR #28197:
URL: https://github.com/apache/camel/pull/28197#discussion_r4333294493
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java:
##########
@@ -134,6 +141,13 @@ public void deliveryComplete(IMqttToken token) {
LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+
+ failedMessageReconnects.set(0);
+ if (stopClient &&
getEndpoint().getConfiguration().isManualAcksEnabled()
Review Comment:
Not blocking: the executor is created after `client.connect(...)` and
`client.subscribe(...)`. With a persistent session, the broker sends the
unacknowledged messages right after CONNACK. A message that fails before this
line finds `reconnectExecutor == null`, `scheduleReconnect` returns false, and
`reconnectAfterFailure` logs at DEBUG that a reconnect is already scheduled,
which is not true. The message then stays unacknowledged until a later failure
triggers a reconnect. Creating the executor before `client.connect(...)` avoids
this.
A related problem is already on main (it could be a follow-up):
`setCallback` is also called after `connect`, so a message the broker sends
again at connect can arrive before there is a callback.
`CommsCallback.messageArrived` then drops it without an acknowledgement.
##########
components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc:
##########
@@ -79,6 +79,24 @@ String payload = "message";
producerTemplate.sendBody("paho-mqtt5:topic", payload);
----
+=== Manual acknowledgements
+
+With `manualAcksEnabled=true` a message is acknowledged when its exchange
completes. MQTT has no negative
+acknowledgement: the broker sends an unacknowledged message again only when
the client connects again to the same
+session, and it does not send more than `receiveMaximum` unacknowledged
messages to the client. So when an exchange
+fails:
+
+* with `cleanStart=false` the consumer disconnects and connects again, and the
broker sends the message again;
Review Comment:
Same as the comment on `rollback`: the broker sends the message again only
for a QoS 1 delivery when `sessionExpiryInterval` is greater than 0. With the
default `qos=2`, Paho has already completed the QoS 2 flow, and without
`sessionExpiryInterval` the session ends at the disconnect. This bullet, the
upgrade guide entry and the `manualAcksEnabled` javadoc should say that.
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java:
##########
@@ -230,6 +250,113 @@ private Exception closeOwnedClient(MqttClient
ownedClient, Exception primaryExce
return primaryException;
}
+ /**
+ * MQTT has no negative acknowledgement: the broker sends an
unacknowledged message again only when the client
+ * connects again to the same session, and it sends no more than
receiveMaximum unacknowledged messages to the
+ * client. A failed message that is just left unacknowledged is therefore
never retried, and enough of them stop the
+ * consumer for good.
+ * <p>
+ * With cleanStart=false the client reconnects, so that the broker sends
the message again; the reconnects of failed
+ * exchanges in a row back off (see {@link #reconnectDelay(int)}), so that
a message that always fails does not
+ * reconnect in a tight loop. With a clean session the broker never sends
it again, so it is acknowledged, which
+ * frees its place in the receive window.
+ */
+ private void rollback(Exchange exchange, MqttMessage mqttMessage) {
+ MqttClient current = client;
+ if (current == null) {
+ return;
+ }
+ if (getEndpoint().getConfiguration().isCleanStart()) {
Review Comment:
Blocking: this sends the failed message to `reconnectAfterFailure` whenever
`cleanStart=false`. The broker sends the message again only in one case: a QoS
1 delivery on a session that outlives the connection.
- QoS 2 (the endpoint default): Paho 1.2.5 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. Manual acks only hold back the QoS 1 PUBACK
(`CommsCallback.handleMessage`). After PUBREC the broker must not send the
PUBLISH again (MQTT 5, 4.3.3), and the PUBCOMP frees the receive window. So a
failed QoS 2 message is neither stuck on main nor sent again here. The
reconnect only disconnects the consumer, sends the other unacknowledged QoS 1
messages again, and logs at WARN that the message will be delivered again.
- QoS 0: there is nothing to acknowledge or to send again.
- `sessionExpiryInterval` not set (Camel default -1, the property is then
not sent): an absent interval means 0 (MQTT 5, 3.1.2.11.2), so the session ends
when the consumer disconnects. The reconnect then loses the failed message, the
other unacknowledged messages, and whatever is published before the subscribe.
On main only the failed message was lost. The new tests set
`sessionExpiryInterval=60`, which hides this.
Hint: reconnect only when `mqttMessage.getQos() == 1`, `cleanStart=false`
and `sessionExpiryInterval > 0` (and `stopClient`). In all other cases call
`messageArrivedComplete` as in the clean session branch. For QoS 2 that also
removes the copy Paho keeps in its persistence until `deliveryComplete`. A test
with `qos=2`, and one with `cleanStart=false` and no `sessionExpiryInterval`,
would cover both.
--
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]
