IllegalArgumentException thrown from PostOffice class

Actual behavior

I'm running Moquette broker and I have 2 client devices that connect to it and publish data once per second. Shortly after the clients connect, I see a large amount of the following exceptions in the logs:

java.util.concurrent.ExecutionException: java.lang.IllegalArgumentException: increment : 0 (expected: > 0)
    at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
    at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191)
    at io.moquette.broker.SessionEventLoop.executeTask(SessionEventLoop.java:49)
    at io.moquette.broker.SessionEventLoop.run(SessionEventLoop.java:34)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.IllegalArgumentException: increment : 0 (expected: > 0)
    at io.netty.util.internal.ObjectUtil.checkPositive(ObjectUtil.java:100)
    at io.netty.util.internal.ReferenceCountUpdater.retain(ReferenceCountUpdater.java:124)
    at io.netty.buffer.AbstractReferenceCountedByteBuf.retain(AbstractReferenceCountedByteBuf.java:86)
    at io.moquette.broker.PostOffice.publish2Subscribers(PostOffice.java:479)   
        at io.moquette.broker.PostOffice.receivedPublishQos1(PostOffice.java:363)
    at io.moquette.broker.MQTTConnection.lambda$processPublish$9(MQTTConnection.java:464)
    at io.moquette.broker.SessionCommand.execute(SessionCommand.java:23)
    at io.moquette.broker.PostOffice.lambda$routeCommand$5(PostOffice.java:635)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at io.moquette.broker.SessionEventLoop.executeTask(SessionEventLoop.java:46)    ... 2 more

This exception is getting thrown like every millisecond or so. Exceptions are originating from this line: https://github.com/moquette-io/moquette/blob/main/broker/src/main/java/io/moquette/broker/PostOffice.java#L472

payload.retain(collector.countBatches());

Apparently collector.countBatches() returns zero, which causes IllegalArgumentException.

Steps to reproduce

The problem is not easy to reproduce. Sometimes the clients connect and publish data without any problems.

Minimal yet complete reproducer code (or URL to code) or complete log file

Attaching AWS Greengrass log. It shows communications between clients and the broker and later the exceptions. greengrass_28.05.log

Moquette MQTT version

Moquette 0.16 running as AWS Greengrass component. It's a version slightly modified by the Greengrass team. Let me know if you think this issue should be reported in their repository instead of here, but it looks more like a Moquette problem. You can check Greengrass's modified version: https://github.com/aws-greengrass/aws-greengrass-moquette-mqtt

JVM version (e.g. `java -version`)

openjdk version "11.0.19" 2023-04-18
OpenJDK Runtime Environment (build 11.0.19+7-post-Ubuntu-0ubuntu118.04.1)
OpenJDK 64-Bit Server VM (build 11.0.19+7-post-Ubuntu-0ubuntu118.04.1, mixed mode, sharing)

OS version (e.g. `uname -a`)

Linu 5.4.0-146-generic #163~18.04.1-Ubuntu SMP Mon Mar 20 15:02:59 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux

moquette-io / moquette