gnodet-bot commented on code in PR #28197:
URL: https://github.com/apache/camel/pull/28197#discussion_r4333524645
##########
components/camel-paho-mqtt5/src/main/docs/paho-mqtt5-component.adoc:
##########
@@ -79,6 +79,29 @@ 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 a session
+that the broker kept, and it does not send more than `receiveMaximum`
unacknowledged messages to the client. The
+client holds back only the acknowledgement of a QoS 1 message; it completes
the QoS 2 flow on its own when the message
+arrives, so the broker never sends a QoS 2 message again. So when an exchange
fails:
+
+* 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, and the broker sends the
+ message again;
+* in all other cases (a clean session, which is the default, a QoS 2 message,
or `cleanStart=false` without
+ `sessionExpiryInterval`, where the session ends with the connection) the
broker never sends the message again, so
+ the consumer acknowledges it, and the failure is left to the Camel error
handler;
+* with a client set by the `client` option, a QoS 1 message on a session that
the broker keeps stays unacknowledged
+ until that client connects again.
+
+The reconnects of failed exchanges in a row back off: the first one waits one
second, each further one waits twice as
Review Comment:
💡 **Minor doc imprecision:** "the first one waits one second" implies a
hardcoded value, but the code actually uses `automaticReconnectMinDelay` from
`MqttConnectionOptions` (defaulting to 1 s). If a user configures a different
min delay, the docs will be misleading. Consider wording like: _"the first one
waits `automaticReconnectMinDelay` (default: one second)"_.
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java:
##########
@@ -230,6 +266,122 @@ 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 a session that the broker kept, and it sends no more
than receiveMaximum unacknowledged
+ * messages to the client. Paho holds back only the PUBACK of a QoS 1
message until it is acknowledged manually; it
+ * completes the QoS 2 flow (PUBREC, PUBCOMP) on its own, so the broker
never sends a QoS 2 message again. A failed
+ * QoS 1 message that is just left unacknowledged is therefore never
retried, and enough of them stop the consumer
+ * for good.
+ * <p>
+ * For a QoS 1 message on a session that the broker keeps after the
disconnect (cleanStart=false and
+ * sessionExpiryInterval greater than 0) 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. In every other case
the broker never sends the message 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 (mqttMessage.getQos() != 1 || !isSessionKeptAfterDisconnect()) {
+ LOG.debug("Acknowledging failed MQTT message with ID: {} (QoS {})
of Exchange ID: {} as the broker does not"
+ + " deliver it again",
+ mqttMessage.getId(), mqttMessage.getQos(),
exchange.getExchangeId());
+ try {
+ current.messageArrivedComplete(mqttMessage.getId(),
mqttMessage.getQos());
+ } catch (MqttException e) {
+ getExceptionHandler().handleException(
+ "Error acknowledging failed MQTT message with ID: " +
mqttMessage.getId(), exchange, e);
+ }
+ } else if (stopClient) {
+ reconnectAfterFailure(current, exchange, mqttMessage);
+ } else {
+ LOG.warn("Rollback due to error processing Exchange ID: {}: MQTT
message with ID: {} stays unacknowledged"
+ + " until the externally provided client connects again",
+ exchange.getExchangeId(), mqttMessage.getId());
+ }
+ }
+
+ private void reconnectAfterFailure(MqttClient current, Exchange exchange,
MqttMessage mqttMessage) {
+ String reason = "Rollback due to error processing Exchange ID: " +
exchange.getExchangeId()
+ + ", so that the broker delivers MQTT message with ID:
" + mqttMessage.getId() + " again";
+ if (!scheduleReconnect(current, reason)) {
+ LOG.debug("{}: a reconnect is already scheduled", reason);
+ }
+ }
+
+ /**
+ * The delay before the reconnect of the given failed exchange in a row:
the minimum automatic reconnect delay of
+ * the client (1 second), doubled for each further failed exchange in a
row, up to maxReconnectDelay.
+ */
+ private long reconnectDelay(int attempt) {
+ long max = Math.max(0,
getEndpoint().getConfiguration().getMaxReconnectDelay());
+ MqttConnectionOptions options = connectionOptions;
+ long min = options != null ?
TimeUnit.SECONDS.toMillis(options.getAutomaticReconnectMinDelay()) : 1000;
+ long delay = min << Math.min(attempt - 1, 20);
+ return Math.min(delay, max);
+ }
Review Comment:
⚠️ **Edge case: `maxReconnectDelay = 0` causes reconnect storm**
When `maxReconnectDelay` is 0 (the default) or negative, `Math.max(0, ...)`
yields `0`, and `Math.min(delay, 0)` clamps every computed delay to **0 ms**.
This means the consumer reconnects immediately in a tight loop, with no backoff
at all — a reconnect storm.
The fix should guard against `max == 0` (i.e. unconfigured) and fall back to
an unbounded delay when the cap is not set:
```suggestion
private long reconnectDelay(int attempt) {
long max = getEndpoint().getConfiguration().getMaxReconnectDelay();
MqttConnectionOptions options = connectionOptions;
long min = options != null ?
TimeUnit.SECONDS.toMillis(options.getAutomaticReconnectMinDelay()) : 1000;
long delay = min << Math.min(attempt - 1, 20);
return max > 0 ? Math.min(delay, max) : delay;
}
```
--
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]
