You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@activemq.apache.org by "Justin Bertram (Jira)" <ji...@apache.org> on 2025/05/12 14:29:00 UTC
[jira] [Updated] (ARTEMIS-5469) MQTT5 flow control rules violation
[ https://issues.apache.org/jira/browse/ARTEMIS-5469?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Justin Bertram updated ARTEMIS-5469:
------------------------------------
Description:
[Section 4.4 of the MQTT5 protocol|https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901238] states:
{quote}
h2. 4.4 Message delivery retry
When a Client reconnects with Clean Start set to 0 and a session is present, both the Client and Server MUST resend any unacknowledged PUBLISH packets (where QoS > 0) and PUBREL packets using their original Packet Identifiers. This is the only circumstance where a Client or Server is REQUIRED to resend messages. Clients and Servers MUST NOT resend messages at any other time [MQTT-4.4.0-1].{quote}
Also, [section 4.9|https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901251] states:
{quote}
4.9 Flow Control
Clients and Servers control the number of unacknowledged PUBLISH packets they receive by using a Receive Maximum value as described in section 3.1.2.11.4 and section 3.2.2.3.2. The Receive Maximum establishes a send quota which is used to limit the number of PUBLISH QOS > 0 packets which can be sent without receiving an PUBACK (for QoS 1) or PUBCOMP (for QoS 2). The PUBACK and PUBCOMP replenish the quota in the manner described below.
The Client or Server MUST set its initial send quota to a non-zero value not exceeding the Receive Maximum [MQTT-4.9.0-1].
Each time the Client or Server sends a PUBLISH packet at QoS > 0, it decrements the send quota. If the send quota reaches zero, the Client or Server MUST NOT send any more PUBLISH packets with QoS > 0 [MQTT-4.9.0-2]. It MAY continue to send PUBLISH packets with QoS 0, or it MAY choose to suspend sending these as well. The Client and Server MUST continue to process and respond to all other MQTT Control Packets even if the quota is zero [MQTT-4.9.0-3].
The send quota is incremented by 1:
- Each time a PUBACK or PUBCOMP packet is received, regardless of whether the PUBACK or PUBCOMP carried an error code.
- Each time a PUBREC packet is received with a Return Code of 0x80 or greater.
The send quota is not incremented if it is already equal to the initial send quota. The attempt to increment above the initial send quota might be caused by the re-transmission of a PUBREL packet after a new Network Connection is established.
*The send quota and Receive Maximum value are not preserved across Network Connections, and are re-initialized with each new Network Connection as described above. They are not part of the session state.*{quote}
But
{code:java}
org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionCallback
public boolean hasCredits(ServerConsumer consumerID, MessageReference ref) {
/*
* [MQTT-3.3.4-9] The Server MUST NOT send more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets for which it
* has not received PUBACK, PUBCOMP, or PUBREC with a Reason Code of 128 or greater from the Client.
*
* Therefore, enforce flow-control based on the number of pending QoS 1 & 2 messages
*/
if (ref != null && ref.isDurable() == true && connection.getReceiveMaximum() != -1 && session.getState().getOutboundStore().getPendingMessages() >= connection.getReceiveMaximum()) {
return false;
} else {
return true;
}
}{code}
uses session object {{OutboundStore}} to determine current pending messages so if
- the client connects with cleanStart=0, session expiry interval>0 and receive maximum = 1
- the broker send PUBLISH and store pending ID to {{OutboundStore}}
- the client disconnects without {{PUBACK}}
- the client reconnects with cleanStart=0, session expiry interval>0 and receive maximum = 1
- the broker will not retry delivery of pending message due to hasCredist==false
was:
Hi!
According to MQTT5 protocol:
[https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html]
{quote}
h2. 4.4 Message delivery retry
When a Client reconnects with Clean Start set to 0 and a session is present, both the Client and Server MUST resend any unacknowledged PUBLISH packets (where QoS > 0) and PUBREL packets using their original Packet Identifiers. This is the only circumstance where a Client or Server is REQUIRED to resend messages. Clients and Servers MUST NOT resend messages at any other time [MQTT-4.4.0-1].
4.9 Flow Control
Clients and Servers control the number of unacknowledged PUBLISH packets they receive by using a Receive Maximum value
The Client or Server MUST set its initial send quota to a non-zero value not exceeding the Receive Maximum [MQTT-4.9.0-1].
Each time the Client or Server sends a PUBLISH packet at QoS > 0, it decrements the send quota. If the send quota reaches zero, the Client or Server MUST NOT send any more PUBLISH packets with QoS > 0 [MQTT-4.9.0-2]. It MAY continue to send PUBLISH packets with QoS 0, or it MAY choose to suspend sending these as well. The Client and Server MUST continue to process and respond to all other MQTT Control Packets even if the quota is zero [MQTT-4.9.0-3].
The send quota is incremented by 1:
· Each time a PUBACK or PUBCOMP packet is received, regardless of whether the PUBACK or PUBCOMP carried an error code.
· Each time a PUBREC packet is received with a Return Code of 0x80 or greater.
The send quota is not incremented if it is already equal to the initial send quota. The attempt to increment above the initial send quota might be caused by the re-transmission of a PUBREL packet after a new Network Connection is established.
*The send quota and Receive Maximum value are not preserved across Network Connections, and are re-initialized with each new Network Connection as described above. They are not part of the session state.*
{quote}
But
{code:java}
org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionCallback
public boolean hasCredits(ServerConsumer consumerID, MessageReference ref) {
/*
* [MQTT-3.3.4-9] The Server MUST NOT send more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets for which it
* has not received PUBACK, PUBCOMP, or PUBREC with a Reason Code of 128 or greater from the Client.
*
* Therefore, enforce flow-control based on the number of pending QoS 1 & 2 messages
*/
if (ref != null && ref.isDurable() == true && connection.getReceiveMaximum() != -1 && session.getState().getOutboundStore().getPendingMessages() >= connection.getReceiveMaximum()) {
return false;
} else {
return true;
}
}{code}
uses session object OutboundStore to determine current pending messages so if
- the client connects with cleanStart=0, session expiry interval>0 and receive maximum = 1
- the broker send PUBLISH and store pending ID to OutboundStore
- the client disconnects without PUBACK
- the client reconnects with cleanStart=0, session expiry interval>0 and receive maximum = 1
- the broker will not retry delivery of pending message due to hasCredist==false
> MQTT5 flow control rules violation
> ----------------------------------
>
> Key: ARTEMIS-5469
> URL: https://issues.apache.org/jira/browse/ARTEMIS-5469
> Project: ActiveMQ Artemis
> Issue Type: Bug
> Components: MQTT
> Reporter: Evgeniy Devyatykh
> Priority: Major
>
> [Section 4.4 of the MQTT5 protocol|https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901238] states:
> {quote}
> h2. 4.4 Message delivery retry
> When a Client reconnects with Clean Start set to 0 and a session is present, both the Client and Server MUST resend any unacknowledged PUBLISH packets (where QoS > 0) and PUBREL packets using their original Packet Identifiers. This is the only circumstance where a Client or Server is REQUIRED to resend messages. Clients and Servers MUST NOT resend messages at any other time [MQTT-4.4.0-1].{quote}
> Also, [section 4.9|https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901251] states:
> {quote}
> 4.9 Flow Control
> Clients and Servers control the number of unacknowledged PUBLISH packets they receive by using a Receive Maximum value as described in section 3.1.2.11.4 and section 3.2.2.3.2. The Receive Maximum establishes a send quota which is used to limit the number of PUBLISH QOS > 0 packets which can be sent without receiving an PUBACK (for QoS 1) or PUBCOMP (for QoS 2). The PUBACK and PUBCOMP replenish the quota in the manner described below.
> The Client or Server MUST set its initial send quota to a non-zero value not exceeding the Receive Maximum [MQTT-4.9.0-1].
> Each time the Client or Server sends a PUBLISH packet at QoS > 0, it decrements the send quota. If the send quota reaches zero, the Client or Server MUST NOT send any more PUBLISH packets with QoS > 0 [MQTT-4.9.0-2]. It MAY continue to send PUBLISH packets with QoS 0, or it MAY choose to suspend sending these as well. The Client and Server MUST continue to process and respond to all other MQTT Control Packets even if the quota is zero [MQTT-4.9.0-3].
> The send quota is incremented by 1:
> - Each time a PUBACK or PUBCOMP packet is received, regardless of whether the PUBACK or PUBCOMP carried an error code.
> - Each time a PUBREC packet is received with a Return Code of 0x80 or greater.
> The send quota is not incremented if it is already equal to the initial send quota. The attempt to increment above the initial send quota might be caused by the re-transmission of a PUBREL packet after a new Network Connection is established.
> *The send quota and Receive Maximum value are not preserved across Network Connections, and are re-initialized with each new Network Connection as described above. They are not part of the session state.*{quote}
> But
> {code:java}
> org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionCallback
> public boolean hasCredits(ServerConsumer consumerID, MessageReference ref) {
> /*
> * [MQTT-3.3.4-9] The Server MUST NOT send more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets for which it
> * has not received PUBACK, PUBCOMP, or PUBREC with a Reason Code of 128 or greater from the Client.
> *
> * Therefore, enforce flow-control based on the number of pending QoS 1 & 2 messages
> */
> if (ref != null && ref.isDurable() == true && connection.getReceiveMaximum() != -1 && session.getState().getOutboundStore().getPendingMessages() >= connection.getReceiveMaximum()) {
> return false;
> } else {
> return true;
> }
> }{code}
> uses session object {{OutboundStore}} to determine current pending messages so if
> - the client connects with cleanStart=0, session expiry interval>0 and receive maximum = 1
> - the broker send PUBLISH and store pending ID to {{OutboundStore}}
> - the client disconnects without {{PUBACK}}
> - the client reconnects with cleanStart=0, session expiry interval>0 and receive maximum = 1
> - the broker will not retry delivery of pending message due to hasCredist==false
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@activemq.apache.org
For additional commands, e-mail: issues-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact