davsclaus commented on code in PR #28197:
URL: https://github.com/apache/camel/pull/28197#discussion_r4333716137
##########
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:
Not blocking, picking up the bot's point below: the default
`maxReconnectDelay` is 128000, so the default case is fine. But an explicit
`maxReconnectDelay` below 1000 (for example 0) makes `Math.min(delay, max)`
return 0, and a message that always fails then reconnects in a tight loop.
Paho's own reconnect never waits less than 1 second. Keeping that floor would
fix it, and `reconnectsAfterFailuresInARowBackOff` (max 1500) still passes:
```suggestion
return Math.max(min, Math.min(delay, max));
```
##########
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()) {
Review Comment:
A question: with a client set through the `client` option, the consumer does
not connect it, so the endpoint's `cleanStart` and `sessionExpiryInterval` may
not match how that client actually connected. If the shared client uses a
persistent session but the endpoint keeps the default `cleanStart=true`, a
failed QoS 1 message is now acknowledged and lost, where before it stayed
unacknowledged and came back on the client's next reconnect. Should the
shared-client case leave the message unacknowledged whatever the endpoint
options say, or should the doc say these options must match the connection
options of the shared client?
--
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]
