Repository: moquette-io/moquette (java)

Issue:
Flow control
## General description

Flow control permit to the client and server to define how many PUBLISH messages with QoS > 0 they could process concurrently per connection. During CONNECT the client uses the `receive maximum` property to communicate to the server how many publishes is can manage concurrently. On the CONNACK the server uses the same property to specify to the client how many publishes the client can send concurrently to the server. If the limit is passed by any of the peers, the offended peer DISCONNECT with reason code 0x93 (Receive Maximum exceeded).


## MQTT5 specification
### 4.9
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](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Receive_Maximum) and [section 3.2.2.3.2](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Receive_Maximum_1). 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.

Refer to [section 3.3.4](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_PUBLISH_Actions) for a description of how Clients and Servers react if they are sent more PUBLISH packets than the Receive Maximum allows.

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.

### 3.1.2.11.4
The Client uses this value to limit the number of QoS 1 and QoS 2 publications that it is willing to process concurrently. There is no mechanism to limit the QoS 0 publications that the Server might try to send.

The value of Receive Maximum applies only to the current Network Connection. If the Receive Maximum value is absent then its value defaults to 65535.

### 3.2.2.3.2
The Server uses this value to limit the number of QoS 1 and QoS 2 publications that it is willing to process concurrently for the Client. It does not provide a mechanism to limit the QoS 0 publications that the Client might try to send.

If the Receive Maximum value is absent, then its value defaults to 65535. 

### 3.3.4
....
`The Client 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 Server [MQTT-3.3.4-7]` f it receives more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets where it has not sent a PUBACK or PUBCOMP in response, the Server uses a DISCONNECT packet with Reason Code 0x93 (Receive Maximum exceeded)
...
`The Client MUST NOT delay the sending of any packets other than PUBLISH packets due to having sent Receive Maximum PUBLISH packets without receiving acknowledgements for them [MQTT-3.3.4-8]` The value of Receive Maximum applies only to the current Network Connection.
...
`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 [MQTT-3.3.4-9].` f it receives more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets where it has not sent a PUBACK or PUBCOMP in response, the Client uses DISCONNECT with Reason Code 0x93 (Receive Maximum exceeded) 
...
`The Server MUST NOT delay the sending of any packets other than PUBLISH packets due to having sent Receive Maximum PUBLISH packets without receiving acknowledgements for them [MQTT-3.3.4-10]. `

## Implementation
### Status
As of today Moquette has a fixed window for output inflight publish messages:
- defines a constant https://github.com/moquette-io/moquette/blob/38428f6c57f981da5c143de38758cc885482907f/broker/src/main/java/io/moquette/BrokerConstants.java#L133
- return in CONNACK's `receive maximum` https://github.com/moquette-io/moquette/blob/38428f6c57f981da5c143de38758cc885482907f/broker/src/main/java/io/moquette/broker/MQTTConnection.java#L369
- use this value as counter limit for inflight in output https://github.com/moquette-io/moquette/blob/38428f6c57f981da5c143de38758cc885482907f/broker/src/main/java/io/moquette/broker/Session.java#L90

#### Problems
The value is hardcoded, can't be defined as a default in the broker configs, it's used to respond to the client to let it know how many concurrent publishes it process but in reality is used to limit the flow in output. 
There isn't any control con the publishes in input.

### Task list
- [ ] read the `receive-maximum` property from CONNECT and use that in the broker output flow to that client, if not specified use the `INFLIGHT_WINDOW_SIZE`. #858
- [x] verify that the counter is incremented and decremented accordingly to the specification. Fixed by #852
- [x] implement a control mechanism to limit the concurrent publishes received from the client, by now use a constant. Once surpassed DISCONNECT with proper reason code, logging the action. Fixed by #852.
- [x] make the incoming `receive maximum` configurable by settings, updating also the fluent config. Fixed by #852

Pull request:
Handled client's receive maximum property to avoid publish flooding
## Release notes
Handled client's receive maximum property to configure the inflight window through the client


## What does this PR do?

The main change contained in this PR is usage of a `sendQuota` in the `MQTTConnection` to keep the inflight zone size instead of `Session`'s `inflightSlots`. The method used to resend inflight messages has been updated to respect the send quota (aka `receive_maximum` property ) expressed by the client on CONNECT message.
If the inflight space is bigger than the actual send quota, maybe because the client squeezed it on a reconnection, it partition the inflight to fill up the send quota. The resend of inflight take precedence over queue drain both on processing of QoS1 PUBACK and QoS2 PURBEC.
If client connects through MQTT 3 the hardcoded limit (10) is used for inflight zone.
If it connects with MQTT5 it reads the `receive_maximum` property or set to default 65535 default value if not specified.

## Why is it important/What is the impact to the user?
Respect the `receive_maximum` asked from the client on MQTT5 connection.

## Checklist
- [x] My code follows the style guidelines of this project
- [x] I have commented my code, particularly in hard-to-understand areas
- ~~[ ] I have made corresponding changes to the documentation~~
- ~~[ ] I have made corresponding change to the default configuration files (and/or docker env variables)~~
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] I have updated the Changelog if it's a feature or a fix that has to be reported

## Related issues

<!-- Recommended
Link related issues below. Insert the issue link or reference after the word "Closes" if merging this should automatically close it.

- Closes #123
- Relates #123
- Requires #123
- Superseeds #123
-->
- Closes #847



Patch:
diff --git a/ChangeLog.txt b/ChangeLog.txt
index f9e059eb..a9ee74ba 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,4 +1,5 @@
 Version 0.18-SNAPSHOT:
+   [feature] Flow-control: Handled client's receive maximum property to configure the inflight window through the client. (#858)
    [feature] Generate correct MANIFEST.MF with bnd-maven-plugin. (#848)
    [feature] Flow-control: implemented publish's quota management on the server side. (#852)
    [fix] Incorrect reference used in compareAndSet in CTrie.cleanTomb. (#841)
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index e61558d2..2b05980d 100644
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -132,6 +132,7 @@ public final class BrokerConstants {
     public static final int FLIGHT_BEFORE_RESEND_MS = 5_000;
     public static final int INFLIGHT_WINDOW_SIZE = 10;
     public static final int INFINITE_SESSION_EXPIRY = 0xFFFFFFFF;
+    public static final int RECEIVE_MAXIMUM = 65 * 1024;
 
     private BrokerConstants() {
     }
diff --git a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
index 210f5b49..2fcced63 100644
--- a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
+++ b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
@@ -66,7 +66,7 @@ class BrokerConfiguration {
             }
         }
 
-        receiveMaximum = props.intProp(IConfig.RECEIVE_MAXIMUM, Integer.MAX_VALUE);
+        receiveMaximum = props.intProp(IConfig.RECEIVE_MAXIMUM, BrokerConstants.RECEIVE_MAXIMUM);
     }
 
     // test method
diff --git a/broker/src/main/java/io/moquette/broker/InflightResender.java b/broker/src/main/java/io/moquette/broker/InflightResender.java
index 514ef5ee..369e9c42 100644
--- a/broker/src/main/java/io/moquette/broker/InflightResender.java
+++ b/broker/src/main/java/io/moquette/broker/InflightResender.java
@@ -55,7 +55,7 @@ public class InflightResender extends ChannelDuplexHandler {
                 // Writer is idle - set a new timeout and notify the callback.
                 resenderTimeout = ctx.executor().schedule(this, resenderTimeNanos, TimeUnit.NANOSECONDS);
                 try {
-                    resendNotAcked(ctx/* , event */);
+                    resendNotAcked(ctx);
                 } catch (Throwable t) {
                     ctx.fireExceptionCaught(t);
                 }
diff --git a/broker/src/main/java/io/moquette/broker/LimitedQuota.java b/broker/src/main/java/io/moquette/broker/LimitedQuota.java
index 4f556ee0..32931fdc 100644
--- a/broker/src/main/java/io/moquette/broker/LimitedQuota.java
+++ b/broker/src/main/java/io/moquette/broker/LimitedQuota.java
@@ -18,8 +18,10 @@
 
 package io.moquette.broker;
 
+import io.moquette.BrokerConstants;
+
 class LimitedQuota implements Quota {
-    private int receiveMaximum;
+    private final int receiveMaximum;
     private int receivedQuota;
 
     public LimitedQuota(int receiveMaximum) {
@@ -29,24 +31,24 @@ class LimitedQuota implements Quota {
 
     @Override
     public boolean hasLimit() {
-        return true;
+        return receiveMaximum != BrokerConstants.RECEIVE_MAXIMUM;
     }
 
     @Override
-    public void decrement() {
+    public void consumeSlot() {
         assert receivedQuota > 0;
         receivedQuota--;
     }
 
     @Override
-    public void increment() {
+    public void releaseSlot() {
         receivedQuota++;
         assert receivedQuota <= receiveMaximum;
     }
 
     @Override
-    public boolean isConsumed() {
-        return receivedQuota == 0;
+    public boolean hasFreeSlots() {
+        return receivedQuota != 0;
     }
 
     @Override
@@ -54,8 +56,13 @@ class LimitedQuota implements Quota {
         return receiveMaximum;
     }
 
+    @Override
+    public int availableSlots() {
+        return receivedQuota;
+    }
+
     @Override
     public String toString() {
-        return "limited quota to " + receivedQuota;
+        return "limited quota to " + receivedQuota + " max slots: " + receiveMaximum;
     }
 }
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
index ceef48ff..7043840e 100644
--- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java
+++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -48,7 +48,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import javax.net.ssl.SSLPeerUnverifiedException;
 
-import static io.moquette.BrokerConstants.INFLIGHT_WINDOW_SIZE;
 import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
 import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
@@ -73,6 +72,7 @@ final class MQTTConnection {
     private Session bindedSession;
     private int protocolVersion;
     private Quota receivedQuota;
+    private Quota sendQuota;
 
     MQTTConnection(Channel channel, BrokerConfiguration brokerConfig, IAuthenticator authenticator,
                    SessionRegistry sessionRegistry, PostOffice postOffice) {
@@ -167,11 +167,11 @@ final class MQTTConnection {
      * Factory method
      * */
     public static Quota createQuota(int receiveMaximum) {
-        if (receiveMaximum == Integer.MAX_VALUE) {
-            return new UnlimitedQuota();
-        } else {
-            return new LimitedQuota(receiveMaximum);
-        }
+        return new LimitedQuota(receiveMaximum);
+    }
+
+    Quota sendQuota() {
+        return sendQuota;
     }
 
     PostOffice.RouteResult processConnect(MqttConnectMessage msg) {
@@ -190,7 +190,7 @@ final class MQTTConnection {
         }
         final boolean cleanSession = msg.variableHeader().isCleanSession();
         final boolean serverGeneratedClientId;
-        if (clientId == null || clientId.length() == 0) {
+        if (clientId == null || clientId.isEmpty()) {
             if (isNotProtocolVersion(msg, MqttVersion.MQTT_5)) {
                 if (!brokerConfig.isAllowZeroByteClientId()) {
                     LOG.info("Broker doesn't permit MQTT empty client ID. Username: {}", username);
@@ -227,6 +227,8 @@ final class MQTTConnection {
         }
         receivedQuota = createQuota(brokerConfig.receiveMaximum());
 
+        sendQuota = retrieveSendQuota(msg);
+
         final String sessionId = clientId;
         protocolVersion = msg.variableHeader().version();
         return postOffice.routeCommand(clientId, "CONN", () -> {
@@ -236,6 +238,28 @@ final class MQTTConnection {
         });
     }
 
+    private Quota retrieveSendQuota(MqttConnectMessage msg) {
+        if (isProtocolVersion(msg, MqttVersion.MQTT_3_1) || isProtocolVersion(msg, MqttVersion.MQTT_3_1_1)) {
+            // for protocol versions that didn't define explicit
+            // the receiver maximum and without specification of flow control
+            // define one by the default.
+            return createQuota(BrokerConstants.INFLIGHT_WINDOW_SIZE);
+        }
+
+        MqttProperties.IntegerProperty receiveMaximumProperty = (MqttProperties.IntegerProperty) msg.variableHeader()
+            .properties()
+            .getProperty(MqttProperties.MqttPropertyType.RECEIVE_MAXIMUM.value());
+        if (receiveMaximumProperty == null) {
+            return createQuota(BrokerConstants.RECEIVE_MAXIMUM);
+        }
+        return createQuota(receiveMaximumProperty.value());
+    }
+
+    // only for test
+    protected void assignSendQuota(Quota quota) {
+        this.sendQuota = quota;
+    }
+
     private void checkMatchSessionLoop(String clientId) {
         if (!sessionLoopDebug) {
             return;
@@ -342,6 +366,7 @@ final class MQTTConnection {
                         LOG.trace("dispatch connection: {}", msg);
                     }
                 } else {
+                    // here the session should still be in CONNECTING state
                     sessionRegistry.connectionClosed(bindedSession);
                     LOG.error("CONNACK send failed, cleanup session and close the connection", future.cause());
                     channel.close();
@@ -395,7 +420,6 @@ final class MQTTConnection {
 
         builder
             .sessionExpiryInterval(BrokerConstants.INFINITE_SESSION_EXPIRY)
-            .receiveMaximum(INFLIGHT_WINDOW_SIZE)
             .retainAvailable(true)
             .wildcardSubscriptionAvailable(true)
             .subscriptionIdentifiersAvailable(true)
@@ -670,8 +694,8 @@ final class MQTTConnection {
                     return null;
                 }).ifFailed(msg::release);
             case AT_LEAST_ONCE:
-                if (receivedQuota.isConsumed()) {
-                    LOG.info("Client {} exceeded the quota {} processing QoS1, disconnecting it", clientId, receivedQuota);
+                if (!receivedQuota.hasFreeSlots()) {
+                    LOG.warn("Client {} exceeded the quota {} processing QoS1, disconnecting it", clientId, receivedQuota);
                     brokerDisconnect(MqttReasonCodes.Disconnect.RECEIVE_MAXIMUM_EXCEEDED);
                     disconnectSession();
                     dropConnection();
@@ -682,16 +706,16 @@ final class MQTTConnection {
                     checkMatchSessionLoop(clientId);
                     if (!isBoundToSession())
                         return null;
-                    receivedQuota.decrement();
+                    receivedQuota.consumeSlot();
                     postOffice.receivedPublishQos1(this, username, messageID, msg, expiry)
                         .completableFuture().thenRun(() -> {
-                            receivedQuota.increment();
+                            receivedQuota.releaseSlot();
                         });
                     return null;
                 }).ifFailed(msg::release);
             case EXACTLY_ONCE: {
-                if (receivedQuota.isConsumed()) {
-                    LOG.info("Client {} exceeded the quota {} processing QoS2, disconnecting it", clientId, receivedQuota);
+                if (!receivedQuota.hasFreeSlots()) {
+                    LOG.warn("Client {} exceeded the quota {} processing QoS2, disconnecting it", clientId, receivedQuota);
                     brokerDisconnect(MqttReasonCodes.Disconnect.RECEIVE_MAXIMUM_EXCEEDED);
                     disconnectSession();
                     dropConnection();
@@ -703,7 +727,7 @@ final class MQTTConnection {
                     if (!isBoundToSession())
                         return null;
                     bindedSession.receivedPublishQos2(messageID, msg);
-                    receivedQuota.decrement();
+                    receivedQuota.consumeSlot();
                     return null;
                 });
                 if (!firstStepResult.isSuccess()) {
@@ -761,7 +785,7 @@ final class MQTTConnection {
             checkMatchSessionLoop(clientID);
             executePubRel(messageID);
             // increment send quota after send PUBCOMP to the client
-            receivedQuota.increment();
+            receivedQuota.releaseSlot();
             return null;
         });
     }
diff --git a/broker/src/main/java/io/moquette/broker/Quota.java b/broker/src/main/java/io/moquette/broker/Quota.java
index 3c8f836d..fcfd99e1 100644
--- a/broker/src/main/java/io/moquette/broker/Quota.java
+++ b/broker/src/main/java/io/moquette/broker/Quota.java
@@ -22,11 +22,13 @@ interface Quota {
 
     boolean hasLimit();
 
-    void decrement();
+    void consumeSlot();
 
-    void increment();
+    void releaseSlot();
 
-    boolean isConsumed();
+    boolean hasFreeSlots();
 
     int getMaximum();
+
+    int availableSlots();
 }
diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java
index df631e63..bd979dfd 100644
--- a/broker/src/main/java/io/moquette/broker/Session.java
+++ b/broker/src/main/java/io/moquette/broker/Session.java
@@ -16,7 +16,7 @@
 package io.moquette.broker;
 
 import static io.moquette.BrokerConstants.FLIGHT_BEFORE_RESEND_MS;
-import static io.moquette.BrokerConstants.INFLIGHT_WINDOW_SIZE;
+
 import io.moquette.broker.SessionRegistry.EnqueuedMessage;
 import io.moquette.broker.SessionRegistry.PublishedMessage;
 import io.moquette.broker.subscriptions.Subscription;
@@ -33,7 +33,6 @@ import java.util.*;
 import java.util.concurrent.DelayQueue;
 import java.util.concurrent.Delayed;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
@@ -44,6 +43,7 @@ class Session {
     // session that doesn't expire, it's ~68 years.
     static final int INFINITE_EXPIRY = Integer.MAX_VALUE;
     private final boolean resendInflightOnTimeout;
+    private Collection<Integer> nonAckPacketIds;
 
     static class InFlightPacket implements Delayed {
 
@@ -87,8 +87,8 @@ class Session {
     // used only in MQTT3 where resends are done on timeout of ACKs.
     private final DelayQueue<InFlightPacket> inflightTimeouts = new DelayQueue<>();
     private final Map<Integer, MqttPublishMessage> qos2Receiving = new HashMap<>();
-    private final AtomicInteger inflightSlots = new AtomicInteger(INFLIGHT_WINDOW_SIZE); // this should be configurable
     private ISessionsRepository.SessionData data;
+    private boolean resendingNonAcked = false;
 
     Session(ISessionsRepository.SessionData data, boolean clean, SessionMessageQueue<SessionRegistry.EnqueuedMessage> sessionQueue) {
         if (sessionQueue == null) {
@@ -214,7 +214,11 @@ class Session {
         MqttMessage pubRel = MQTTConnection.pubrel(pubRecPacketId);
         mqttConnection.sendIfWritableElseDrop(pubRel);
 
-        drainQueueToConnection();
+        if (resendingNonAcked) {
+            resendInflightNotAcked();
+        } else {
+            drainQueueToConnection();
+        }
     }
 
     public void processPubComp(int messageID) {
@@ -222,11 +226,11 @@ class Session {
         cleanFromInflight(messageID);
         SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(messageID);
         if (removed == null) {
-            LOG.warn("Received a PUBCOMP with not matching packetId");
+            LOG.warn("Received a PUBCOMP with not matching packetId in the inflight cache");
             return;
         }
         removed.release();
-        inflightSlots.incrementAndGet();
+        mqttConnection.sendQuota().releaseSlot();
         drainQueueToConnection();
 
         // TODO notify the interceptor
@@ -301,7 +305,7 @@ class Session {
         publishRequest.retain();
 
         if (canSkipQueue(localMqttConnectionRef)) {
-            inflightSlots.decrementAndGet();
+            mqttConnection.sendQuota().consumeSlot();
             int packetId = localMqttConnectionRef.nextPacketId();
 
             LOG.debug("Adding into inflight for session {} at QoS {}", getClientID(), publishRequest.getPublishingQos());
@@ -310,7 +314,7 @@ class Session {
             // If there already was something, release it.
             if (old != null) {
                 old.release();
-                inflightSlots.incrementAndGet();
+                mqttConnection.sendQuota().releaseSlot();
             }
             if (resendInflightOnTimeout) {
                 inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
@@ -332,13 +336,13 @@ class Session {
     private boolean canSkipQueue(MQTTConnection localMqttConnectionRef) {
         return localMqttConnectionRef != null &&
             sessionQueue.isEmpty() &&
-            inflightSlots.get() > 0 &&
+            mqttConnection.sendQuota().hasFreeSlots() &&
             connected() &&
             localMqttConnectionRef.channel.isWritable();
     }
 
     private boolean inflightHasSlotsAndConnectionIsUp() {
-        return inflightSlots.get() > 0 &&
+        return mqttConnection.sendQuota().hasFreeSlots() &&
             connected() &&
             mqttConnection.channel.isWritable();
     }
@@ -348,14 +352,19 @@ class Session {
         cleanFromInflight(ackPacketId);
         SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(ackPacketId);
         if (removed == null) {
-            LOG.warn("Received a PUBACK with not matching packetId");
+            LOG.warn("Received a PUBACK with not matching packetId({}) in the inflight cache({})",
+                ackPacketId, inflightWindow.keySet());
             return;
         }
         removed.release();
 
-        inflightSlots.incrementAndGet();
+        mqttConnection.sendQuota().releaseSlot();
         LOG.debug("Received PUBACK {} for session {}", ackPacketId, getClientID());
-        drainQueueToConnection();
+        if (resendingNonAcked) {
+            resendInflightNotAcked();
+        } else {
+            drainQueueToConnection();
+        }
     }
 
     private void cleanFromInflight(int ackPacketId) {
@@ -367,21 +376,45 @@ class Session {
     }
 
     public void resendInflightNotAcked() {
-        Collection<Integer> nonAckPacketIds;
-        if (resendInflightOnTimeout) {
-            // MQTT3 behavior, resend on timeout
-            Collection<InFlightPacket> expired = new ArrayList<>(INFLIGHT_WINDOW_SIZE);
-            inflightTimeouts.drainTo(expired);
-            nonAckPacketIds = expired.stream().map(p -> p.packetId).collect(Collectors.toList());
+        if (!resendingNonAcked) {
+            if (resendInflightOnTimeout) {
+                // MQTT3 behavior, resend on timeout
+                Collection<InFlightPacket> expired = new ArrayList<>();
+                inflightTimeouts.drainTo(expired);
+                nonAckPacketIds = expired.stream().map(p -> p.packetId).collect(Collectors.toList());
+            } else {
+                // MQTT5 behavior resend only not acked present in reopened session.
+                // need a copy else removing from the nonAckPacketIds would remove also from inflightWindow
+                nonAckPacketIds = new ArrayList<>(inflightWindow.keySet());
+            }
+
+            debugLogPacketIds(nonAckPacketIds);
+        }
+
+        if (nonAckPacketIds.size() > mqttConnection.sendQuota().availableSlots()) {
+            // Send quota is smaller than the inflight messages to resend, split it.
+            // Next partition will be sent on PUBREC or PUBACK reception to continue flushing the in flight.
+            resendingNonAcked = true;
+            List<Integer> partition = nonAckPacketIds.stream()
+                .limit(mqttConnection.sendQuota().availableSlots())
+                .collect(Collectors.toList());
+
+            resendNonAckedIdsPartition(partition);
+
+            // clean up the partition sent
+            for (Integer id : partition) {
+                nonAckPacketIds.remove(id);
+            }
         } else {
-            // MQTT5 behavior resend only not acked present in reopened session.
-            nonAckPacketIds = inflightWindow.keySet();
+            resendNonAckedIdsPartition(nonAckPacketIds);
+            resendingNonAcked = false;
         }
 
-        debugLogPacketIds(nonAckPacketIds);
+    }
 
-        for (Integer notAckPacketId : nonAckPacketIds) {
-            final SessionRegistry.EnqueuedMessage msg = inflightWindow.get(notAckPacketId);
+    private void resendNonAckedIdsPartition(Collection<Integer> packetIdsToResend) {
+        for (Integer notAckPacketId : packetIdsToResend) {
+            final EnqueuedMessage msg = inflightWindow.get(notAckPacketId);
             if (msg == null) {
                 // Already acked...
                 continue;
@@ -393,7 +426,7 @@ class Session {
                 }
                 mqttConnection.sendIfWritableElseDrop(pubRel);
             } else {
-                final SessionRegistry.PublishedMessage pubMsg = (SessionRegistry.PublishedMessage) msg;
+                final PublishedMessage pubMsg = (PublishedMessage) msg;
                 final Topic topic = pubMsg.topic;
                 final MqttQoS qos = pubMsg.publishingQos;
                 final ByteBuf payload = pubMsg.payload;
@@ -405,6 +438,8 @@ class Session {
                     inflightTimeouts.add(new InFlightPacket(notAckPacketId, FLIGHT_BEFORE_RESEND_MS));
                 }
                 mqttConnection.sendPublish(publishMsg);
+
+                mqttConnection.sendQuota().consumeSlot();
             }
         }
     }
@@ -435,14 +470,14 @@ class Session {
                 return;
             }
 
-            inflightSlots.decrementAndGet();
+            mqttConnection.sendQuota().consumeSlot();
             int sendPacketId = mqttConnection.nextPacketId();
 
             // Putting it in a map, but the retain is cancelled out by the below release.
             EnqueuedMessage old = inflightWindow.put(sendPacketId, msg);
             if (old != null) {
                 old.release();
-                inflightSlots.incrementAndGet();
+                mqttConnection.sendQuota().releaseSlot();
             }
             if (resendInflightOnTimeout) {
                 inflightTimeouts.add(new InFlightPacket(sendPacketId, FLIGHT_BEFORE_RESEND_MS));
@@ -470,8 +505,12 @@ class Session {
         LOG.trace("Republishing all saved messages for session {}", this);
         resendInflightNotAcked();
 
-        // send queued messages while offline
-        drainQueueToConnection();
+        if (!resendingNonAcked) {
+            // if resend of inflight is bigger than send quota, till it's finished
+            // do not drain the queue.
+            // send queued messages while offline
+            drainQueueToConnection();
+        }
     }
 
     public void receivedPublishQos2(int messageID, MqttPublishMessage msg) {
@@ -529,7 +568,7 @@ class Session {
             "clientId='" + data.clientId() + '\'' +
             ", clean=" + clean +
             ", status=" + status +
-            ", inflightSlots=" + inflightSlots +
+            ", inflightSlots=" + mqttConnection.sendQuota().availableSlots() +
             '}';
     }
 }
diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
index e6520971..b1049a7d 100644
--- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java
+++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
@@ -38,6 +38,7 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
 
+import static io.moquette.BrokerConstants.INFLIGHT_WINDOW_SIZE;
 import static io.moquette.broker.Session.INFINITE_EXPIRY;
 
 public class SessionRegistry {
diff --git a/broker/src/main/java/io/moquette/broker/UnlimitedQuota.java b/broker/src/main/java/io/moquette/broker/UnlimitedQuota.java
deleted file mode 100644
index 42ec1f25..00000000
--- a/broker/src/main/java/io/moquette/broker/UnlimitedQuota.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- *
- * Copyright (c) 2012-2024 The original author or authors
- * ------------------------------------------------------
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
- * 
- * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
- * 
- * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
- * 
- * You may elect to redistribute this code under either of these licenses.
- *
- */
-
-package io.moquette.broker;
-
-class UnlimitedQuota implements Quota {
-
-    @Override
-    public boolean hasLimit() {
-        return false;
-    }
-
-    @Override
-    public void decrement() {
-        // do nothing for unlimited quota
-    }
-
-    @Override
-    public void increment() {
-        // do nothing for unlimited quota
-    }
-
-    @Override
-    public boolean isConsumed() {
-        return false;
-    }
-
-    @Override
-    public int getMaximum() {
-        return Integer.MAX_VALUE;
-    }
-
-    @Override
-    public String toString() {
-        return "unlimited quota";
-    }
-}


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/SessionTest.java b/broker/src/test/java/io/moquette/broker/SessionTest.java
index cf512c67..99fc6b22 100644
--- a/broker/src/test/java/io/moquette/broker/SessionTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionTest.java
@@ -11,7 +11,6 @@ import io.netty.handler.codec.mqtt.MqttVersion;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import static io.moquette.BrokerConstants.FLIGHT_BEFORE_RESEND_MS;
 import io.moquette.broker.subscriptions.Subscription;
 
 import java.time.Clock;
@@ -19,10 +18,10 @@ import java.time.Instant;
 import java.util.Arrays;
 import org.assertj.core.api.Assertions;
 
+import static io.moquette.BrokerConstants.*;
 import static io.moquette.broker.Session.INFINITE_EXPIRY;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 
 public class SessionTest {
 
@@ -128,6 +127,7 @@ public class SessionTest {
     private void createConnection(Session client) {
         BrokerConfiguration brokerConfiguration = new BrokerConfiguration(true, false, false, NO_BUFFER_FLUSH);
         MQTTConnection mqttConnection = new MQTTConnection(testChannel, brokerConfiguration, null, null, null);
+        mqttConnection.assignSendQuota(new LimitedQuota(INFLIGHT_WINDOW_SIZE));
         client.markConnecting();
         client.bind(mqttConnection);
         client.completeConnection();
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java
index e18709f3..8dd989d0 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java
@@ -1,18 +1,30 @@
 package io.moquette.integration.mqtt5;
 
 import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import io.moquette.BrokerConstants;
 import io.moquette.testclient.Client;
-import io.netty.handler.codec.mqtt.MqttConnAckMessage;
+import io.netty.handler.codec.mqtt.*;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+
 import static io.moquette.integration.mqtt5.TestUtils.assertConnectionAccepted;
+import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public abstract class AbstractServerIntegrationTest extends AbstractServerIntegrationWithoutClientFixture {
 
     Client lowLevelClient;
 
+    static void sleepSeconds(int secondsInterval) throws InterruptedException {
+        Thread.sleep(Duration.ofSeconds(secondsInterval).toMillis());
+    }
+
     @NotNull
     Mqtt5BlockingClient createSubscriberClient() {
         String clientId = clientName();
@@ -40,7 +52,33 @@ public abstract class AbstractServerIntegrationTest extends AbstractServerIntegr
     }
 
     void connectLowLevel(int keepAliveSecs) {
-        MqttConnAckMessage connAck = lowLevelClient.connectV5(keepAliveSecs);
+        MqttConnAckMessage connAck = lowLevelClient.connectV5(keepAliveSecs, BrokerConstants.INFLIGHT_WINDOW_SIZE);
         assertConnectionAccepted(connAck, "Connection must be accepted");
     }
+
+    protected void consumesPublishesInflightWindow(int inflightWindowSize) throws InterruptedException {
+        for (int i = 0; i < inflightWindowSize; i++) {
+            consumePendingPublishAndAcknowledge(Integer.toString(i));
+        }
+    }
+
+    protected void consumePendingPublishAndAcknowledge(String expectedPayload) throws InterruptedException {
+        MqttMessage mqttMessage = lowLevelClient.receiveNextMessage(Duration.ofMillis(20000));
+        assertNotNull(mqttMessage, "A message MUST be received");
+
+        assertEquals(MqttMessageType.PUBLISH, mqttMessage.fixedHeader().messageType(), "Message received should MqttPublishMessage");
+        MqttPublishMessage publish = (MqttPublishMessage) mqttMessage;
+        assertEquals(expectedPayload, publish.payload().toString(StandardCharsets.UTF_8));
+        int packetId = publish.variableHeader().packetId();
+        assertTrue(publish.release(), "Reference of publish should be released");
+
+        acknowledge(packetId);
+    }
+
+    private void acknowledge(int packetId) {
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE,
+            false, 0);
+        MqttPubAckMessage pubAck = new MqttPubAckMessage(fixedHeader, MqttMessageIdVariableHeader.from(packetId));
+        lowLevelClient.sendMessage(pubAck);
+    }
 }
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java
index 24be7238..12b3479c 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java
@@ -45,7 +45,6 @@ class ConnectAckTest extends  AbstractServerIntegrationTest {
     public void testAckResponseProperties() {
         final MqttProperties ackProps = connAck.variableHeader().properties();
         verifyProperty(MqttPropertyType.SESSION_EXPIRY_INTERVAL, ackProps, BrokerConstants.INFINITE_SESSION_EXPIRY, "Session expiry is infinite");
-        verifyProperty(MqttPropertyType.RECEIVE_MAXIMUM, ackProps, INFLIGHT_WINDOW_SIZE, "Receive maximum property must equals flight window size");
         verifyNotSet(MqttPropertyType.MAXIMUM_QOS, ackProps, "Maximum QoS is not set => QoS 2 ready");
         verifyProperty(MqttPropertyType.RETAIN_AVAILABLE, ackProps, 1, "Retain feature is available");
         verifyNotSet(MqttPropertyType.MAXIMUM_PACKET_SIZE, ackProps, "Maximum packet size is the one defined by specs");
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/FlowControlTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/FlowControlTest.java
index 78f9f129..7df84f54 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/FlowControlTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/FlowControlTest.java
@@ -18,6 +18,8 @@
 
 package io.moquette.integration.mqtt5;
 
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
 import io.moquette.broker.config.FluentConfig;
 import io.moquette.broker.config.IConfig;
 import io.moquette.testclient.Client;
@@ -28,11 +30,13 @@ import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 
 import static io.moquette.integration.mqtt5.TestUtils.assertConnectionAccepted;
 import static io.netty.handler.codec.mqtt.MqttQoS.EXACTLY_ONCE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class FlowControlTest extends AbstractServerIntegrationTest {
@@ -97,4 +101,87 @@ public class FlowControlTest extends AbstractServerIntegrationTest {
     public String clientName() {
         return "sender";
     }
+
+
+    @Test
+    public void givenClientConnectedWithCertainReceiveMaximumWhenInFlightSizeIsSurpassedThenTheServerEnqueueAndDontFloodTheClient() throws InterruptedException {
+        connectLowLevel();
+
+        // subscribe with an identifier
+        MqttMessage received = lowLevelClient.subscribeWithIdentifier("temperature/living",
+            MqttQoS.AT_LEAST_ONCE, 123);
+        verifyOfType(received, MqttMessageType.SUBACK);
+
+        //lowlevel client doesn't ACK any pub, so the in flight window fills up
+        Mqtt5BlockingClient publisher = createPublisherClient();
+        int inflightWindowSize = 10;
+        // fill the in flight window so that messages starts to be enqueued
+        fillInFlightWindow(inflightWindowSize, publisher);
+
+        // send another message, which is enqueued and has an expiry of messageExpiryInterval seconds
+        publisher.publishWith()
+            .topic("temperature/living")
+            .payload(("Enqueued").getBytes(StandardCharsets.UTF_8))
+            .qos(MqttQos.AT_LEAST_ONCE) // Broker enqueues only QoS1 and QoS2001
+            .send();
+
+        // after sending more publish the receive maximum limit is not passed and the connection remain open
+        assertTrue(lowLevelClient.isConnected());
+
+        // now subscriber consumes messages, shouldn't receive any message in the form "Enqueued-"
+        consumesPublishesInflightWindow(inflightWindowSize);
+
+        MqttMessage mqttMessage = lowLevelClient.receiveNextMessage(Duration.ofMillis(100));
+        assertNotNull(mqttMessage, "A message MUST be received");
+
+        assertEquals(MqttMessageType.PUBLISH, mqttMessage.fixedHeader().messageType(), "Message received should MqttPublishMessage");
+        MqttPublishMessage publish = (MqttPublishMessage) mqttMessage;
+        assertEquals("Enqueued", publish.payload().toString(StandardCharsets.UTF_8));
+        assertTrue(publish.release(), "Reference of publish should be released");
+    }
+
+    private static void fillInFlightWindow(int numPublishToSend, Mqtt5BlockingClient publisher) {
+        for (int i = 0; i < numPublishToSend; i++) {
+            publisher.publishWith()
+                .topic("temperature/living")
+                .payload(Integer.toString(i).getBytes(StandardCharsets.UTF_8))
+                .qos(MqttQos.AT_LEAST_ONCE)
+                .send();
+        }
+    }
+
+    @Test
+    public void givenClientThatReconnectWithSmallerReceiveMaximumThenForwardCorrectlyTheFullListOfPendingMessagesWithoutAnyLose() throws InterruptedException {
+        // connect subscriber and published
+        // publisher send 20 events, 10 should be in the inflight, 10 remains on the queue
+        connectLowLevel();
+
+        // subscribe with an identifier
+        MqttMessage received = lowLevelClient.subscribeWithIdentifier("temperature/living",
+            MqttQoS.AT_LEAST_ONCE, 123);
+        verifyOfType(received, MqttMessageType.SUBACK);
+
+        //lowlevel client doesn't ACK any pub, so the in flight window fills up
+        Mqtt5BlockingClient publisher = createPublisherClient();
+        int inflightWindowSize = 10;
+        // fill the in flight window so that messages starts to be enqueued
+        fillInFlightWindow(inflightWindowSize + 10, publisher);
+
+        System.out.println("Filled inflight and queue");
+
+        // disconnect subscriber
+        lowLevelClient.disconnect();
+        lowLevelClient.close();
+
+        System.out.println("Closed old client, reconnecting");
+
+        // reconnect the subscriber with smaller received maximum
+        lowLevelClient = new Client("localhost").clientId(clientName());
+        MqttConnAckMessage connAck = lowLevelClient.connectV5WithReceiveMaximum(5);
+        assertConnectionAccepted(connAck, "Connection must be re-accepted with smaller window size");
+        System.out.println("Client reconnected second time");
+
+        // should receive all the 20 messages
+        consumesPublishesInflightWindow(inflightWindowSize + 10);
+    }
 }
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/MessageExpirationTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/MessageExpirationTest.java
index 20715f10..14ab7119 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/MessageExpirationTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/MessageExpirationTest.java
@@ -147,21 +147,21 @@ public class MessageExpirationTest extends AbstractServerIntegrationTest {
     }
 
     @Test
-    public void givenPublishedMessageWithExpiryWhenMessageRemainInBrokerForMoreThanTheExipiryIsNotPublished() throws InterruptedException {
-        int messageExpiryInterval = 2; // seconds
+    public void givenPublishWithMessageExpiryPropertyWhenItsForwardedToSubscriberThenExpiryValueHasToBeDeducedByTheTimeSpentInBroker() throws InterruptedException {
+        int messageExpiryInterval = 10; // seconds
         // avoid the keep alive period could disconnect
-        connectLowLevel(messageExpiryInterval * 2);
+        connectLowLevel((int) (messageExpiryInterval * 1.5));
 
         // subscribe with an identifier
         MqttMessage received = lowLevelClient.subscribeWithIdentifier("temperature/living",
-            MqttQoS.AT_LEAST_ONCE, 123);
+            MqttQoS.AT_LEAST_ONCE, 123, 500, TimeUnit.MILLISECONDS);
         verifyOfType(received, MqttMessageType.SUBACK);
 
         //lowlevel client doesn't ACK any pub, so the in flight window fills up
         Mqtt5BlockingClient publisher = createPublisherClient();
         int inflightWindowSize = 10;
         // fill the in flight window so that messages starts to be enqueued
-        fillInFlightWindow(inflightWindowSize, publisher, messageExpiryInterval);
+        fillInFlightWindow(inflightWindowSize, publisher, Integer.MIN_VALUE);
 
         // send another message, which is enqueued and has an expiry of messageExpiryInterval seconds
         publisher.publishWith()
@@ -172,31 +172,44 @@ public class MessageExpirationTest extends AbstractServerIntegrationTest {
             .send();
 
         // let time flow so that the message in queue passes its expiry time
-        Thread.sleep(Duration.ofSeconds(messageExpiryInterval + 1).toMillis());
+        long sleepMillis = Duration.ofSeconds(messageExpiryInterval / 2).toMillis();
+        Thread.sleep(sleepMillis);
 
         // now subscriber consumes messages, shouldn't receive any message in the form "Enqueued-"
         consumesPublishesInflightWindow(inflightWindowSize);
+        MqttMessage mqttMessage = lowLevelClient.receiveNextMessage(Duration.ofMillis(1000));
 
-        MqttMessage mqttMessage = lowLevelClient.receiveNextMessage(Duration.ofMillis(100));
-        assertNull(mqttMessage, "No other messages MUST be received after consuming the in flight window");
+        assertNotNull(mqttMessage, "A publish out of the queue has to be received");
+        assertEquals(MqttMessageType.PUBLISH, mqttMessage.fixedHeader().messageType(), "Expected a publish message");
+        MqttPublishMessage publishMessage = (MqttPublishMessage) mqttMessage;
+
+        // extract message expiry property
+        MqttProperties.MqttProperty expiryProp = publishMessage.variableHeader()
+            .properties()
+            .getProperty(MqttProperties.MqttPropertyType.PUBLICATION_EXPIRY_INTERVAL.value());
+        assertNotNull(expiryProp, "Publication expiry property can't be null");
+        Integer expirySeconds = ((MqttProperties.IntegerProperty) expiryProp).value();
+
+        assertTrue(expirySeconds < messageExpiryInterval, "Publish's expiry has to be updated");
+        assertTrue(publishMessage.release(), "Last reference of publish should be released");
     }
 
     @Test
-    public void givenPublishWithMessageExpiryPropertyWhenItsForwardedToSubscriberThenExpiryValueHasToBeDeducedByTheTimeSpentInBroker() throws InterruptedException {
-        int messageExpiryInterval = 10; // seconds
+    public void givenPublishedMessageWithExpiryWhenMessageRemainInBrokerForMoreThanTheExpiryIsNotPublished() throws InterruptedException {
+        int messageExpiryInterval = 2; // seconds
         // avoid the keep alive period could disconnect
-        connectLowLevel((int) (messageExpiryInterval * 1.5));
+        connectLowLevel(messageExpiryInterval * 2);
 
         // subscribe with an identifier
         MqttMessage received = lowLevelClient.subscribeWithIdentifier("temperature/living",
-            MqttQoS.AT_LEAST_ONCE, 123, 500, TimeUnit.MILLISECONDS);
+            MqttQoS.AT_LEAST_ONCE, 123);
         verifyOfType(received, MqttMessageType.SUBACK);
 
         //lowlevel client doesn't ACK any pub, so the in flight window fills up
         Mqtt5BlockingClient publisher = createPublisherClient();
         int inflightWindowSize = 10;
         // fill the in flight window so that messages starts to be enqueued
-        fillInFlightWindow(inflightWindowSize, publisher, Integer.MIN_VALUE);
+        fillInFlightWindow(inflightWindowSize, publisher, messageExpiryInterval);
 
         // send another message, which is enqueued and has an expiry of messageExpiryInterval seconds
         publisher.publishWith()
@@ -207,44 +220,13 @@ public class MessageExpirationTest extends AbstractServerIntegrationTest {
             .send();
 
         // let time flow so that the message in queue passes its expiry time
-        long sleepMillis = Duration.ofSeconds(messageExpiryInterval / 2).toMillis();
-        Thread.sleep(sleepMillis);
+        sleepSeconds(messageExpiryInterval + 1);
 
         // now subscriber consumes messages, shouldn't receive any message in the form "Enqueued-"
         consumesPublishesInflightWindow(inflightWindowSize);
-        MqttMessage mqttMessage = lowLevelClient.receiveNextMessage(Duration.ofMillis(1000));
 
-        assertNotNull(mqttMessage, "A publish out of the queue has to be received");
-        assertEquals(MqttMessageType.PUBLISH, mqttMessage.fixedHeader().messageType(), "Expected a publish message");
-        MqttPublishMessage publishMessage = (MqttPublishMessage) mqttMessage;
-
-        // extract message expiry property
-        MqttProperties.MqttProperty expiryProp = publishMessage.variableHeader()
-            .properties()
-            .getProperty(MqttProperties.MqttPropertyType.PUBLICATION_EXPIRY_INTERVAL.value());
-        assertNotNull(expiryProp, "Publication expiry property can't be null");
-        Integer expirySeconds = ((MqttProperties.IntegerProperty) expiryProp).value();
-
-        assertTrue(expirySeconds < messageExpiryInterval, "Publish's expiry has to be updated");
-        assertTrue(publishMessage.release(), "Last reference of publish should be released");
-    }
-
-    private void consumesPublishesInflightWindow(int inflightWindowSize) throws InterruptedException {
-        for (int i = 0; i < inflightWindowSize; i++) {
-            MqttMessage mqttMessage = lowLevelClient.receiveNextMessage(Duration.ofMillis(20000));
-            assertNotNull(mqttMessage, "A message MUST be received");
-
-            assertEquals(MqttMessageType.PUBLISH, mqttMessage.fixedHeader().messageType(), "Message received should MqttPublishMessage");
-            MqttPublishMessage publish = (MqttPublishMessage) mqttMessage;
-            assertEquals(Integer.toString(i), publish.payload().toString(StandardCharsets.UTF_8));
-            int packetId = publish.variableHeader().packetId();
-            assertTrue(publish.release(), "Reference of publish should be released");
-
-            MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE,
-                false, 0);
-            MqttPubAckMessage pubAck = new MqttPubAckMessage(fixedHeader, MqttMessageIdVariableHeader.from(packetId));
-            lowLevelClient.sendMessage(pubAck);
-        }
+        MqttMessage mqttMessage = lowLevelClient.receiveNextMessage(Duration.ofMillis(100));
+        assertNull(mqttMessage, "No other messages MUST be received after consuming the in flight window");
     }
 
     private static void fillInFlightWindow(int inflightWindowSize, Mqtt5BlockingClient publisher, int messageExpiryInterval) {
diff --git a/broker/src/test/java/io/moquette/testclient/Client.java b/broker/src/test/java/io/moquette/testclient/Client.java
index ba8e8427..b406bcf3 100644
--- a/broker/src/test/java/io/moquette/testclient/Client.java
+++ b/broker/src/test/java/io/moquette/testclient/Client.java
@@ -139,19 +139,31 @@ public class Client {
     }
 
     public MqttConnAckMessage connectV5() {
-        return connectV5(2);
+        return connectV5(2, BrokerConstants.INFLIGHT_WINDOW_SIZE);
+    }
+
+    public MqttConnAckMessage connectV5WithReceiveMaximum(int receiveMaximumInflight) {
+        return connectV5(2, receiveMaximumInflight);
     }
 
     @NotNull
-    public MqttConnAckMessage connectV5(int keepAliveSecs) {
+    public MqttConnAckMessage connectV5(int keepAliveSecs, int receiveMaximumInflight) {
         final MqttMessageBuilders.ConnectBuilder builder = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_5);
         if (clientId != null) {
             builder.clientId(clientId);
         }
+
+        final MqttProperties connectProperties = new MqttProperties();
+        MqttProperties.IntegerProperty receiveMaximum = new MqttProperties.IntegerProperty(
+            MqttProperties.MqttPropertyType.RECEIVE_MAXIMUM.value(),
+            receiveMaximumInflight);
+        connectProperties.add(receiveMaximum);
+
         MqttConnectMessage connectMessage = builder
             .keepAlive(keepAliveSecs) // secs
             .willFlag(false)
             .willQoS(MqttQoS.AT_MOST_ONCE)
+            .properties(connectProperties)
             .build();
 
         return doConnect(connectMessage);
@@ -367,4 +379,8 @@ public class Client {
         }
         workerGroup.shutdownGracefully();
     }
+
+    public boolean isConnected() {
+        return m_channel.isActive();
+    }
 }