fix: deliver mqtt messages at granted qos with per-subscriber packet id by wy471x · Pull Request #6913 · apache/shenyu

wy471x · 2026-08-13T15:54:02Z

Store the granted qos per subscriber channel in SubscribeRepository and allocate a packet id from each subscriber's own id space when fanning out publishes, instead of hard-coding AT_MOST_ONCE and reusing the publisher's packet id for every subscriber.

Make sure that:

You have read the contribution guidelines.
You submit test cases (unit or integration tests) that back your changes.
Your local test passed ./mvnw clean install -Dmaven.javadoc.skip=true.

Summary

Fixed MQTT broker→subscriber fan-out to respect granted QoS and per-subscriber packet IDs.

Changes

SubscribeRepository — now stores topic → Map<Channel, MqttQoS> instead of topic → List; keeps the max QoS for overlapping subscriptions and ignores FAILURE
subscriptions.
MqttPacketIdGenerator (new) — allocates packet IDs 1–65535 (wrapping) from each channel's own ID space; state cleaned up on Disconnect.
Publish.send — delivers at min(publishQoS, grantedQoS), assigns a per-subscriber packet ID for QoS > 0, and uses 0 for QoS 0, instead of hard-coding AT_MOST_ONCE and reusing the
publisher's packet ID.

Tests — 13 new unit tests covering the generator, repository, and publish fan-out behavior; all pass with checkstyle.

close #6850

New Changs

#6743 — QoS 2 PUBLISH answered with PUBACK instead of PUBREC

Publish.java

qos2() now responds with PUBREC instead of PUBACK — the first ack of the QoS 2 four-step handshake
Fixed reserved fixed-header bits on both acks: PUBACK/PUBREC must encode QoS bits as 00 (AT_MOST_ONCE); previously qos1 sent AT_LEAST_ONCE and qos2 sent EXACTLY_ONCE, which
spec-compliant clients treat as a protocol error
Removed the unused empty qos0() stub

PUBREL → PUBCOMP half-handshake (new)

New PubRel class: on receiving PUBREL, responds with PUBCOMP echoing the same packet id (flags 0000)
Wired the dispatch chain: AbstractMessageType.pubRel() default method → MessageType.pubRel() delegate → MqttFactory PUBREL case

Tests: testPublishQos2SendsPubRecToPublisher verifies PUBREC type/flags/message-id; new PubRelTest verifies PUBCOMP echo.

#6744 — No cleanup on channel close → repository leak

MqttTransportHandler.java

Implemented the previously empty operationComplete(): removes the closing channel from ChannelRepository, SubscribeRepository, and MqttPacketIdGenerator
The handler is already registered as a listener on ch.closeFuture(), so both graceful and ungraceful (TCP drop) disconnects now trigger cleanup

SubscribeRepository.java

Added remove(Channel) overload that removes the channel from every topic it subscribed to (existing APIs only removed by topic)

Test: new MqttTransportHandlerTest uses an EmbeddedChannel to verify all three repositories no longer hold the channel after close, and the packet-id space resets.

close #6743
close #6744

Store the granted qos per subscriber channel in SubscribeRepository and allocate a packet id from each subscriber's own id space when fanning out publishes, instead of hard-coding AT_MOST_ONCE and reusing the publisher's packet id for every subscriber. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Aias00

This is a solid fix that brings the MQTT fan-out into line with the spec: it now delivers at min(published QoS, granted QoS) and allocates a fresh packet identifier from each subscriber channel's own id space, instead of hard-coding AT_MOST_ONCE and reusing the publisher's packet id. The change is well-scoped and backed by thorough unit tests (PublishTest, SubscribeRepositoryTest, MqttPacketIdGeneratorTest) covering granted-QoS fan-out, QoS0 → packetId 0, per-subscriber id spaces, max-QoS merge, FAILURE filtering, and id wrap-around at 0xFFFF.

A few non-blocking notes:

Dead code: int packetId = msg.variableHeader().packetId(); in Publish.publish is now unused (it used to be passed to send). Please remove it, along with the now-resolved //// todo qos comment.
Id-space cleanup: MqttPacketIdGenerator.CHANNEL_PACKET_ID_FACTORY is a static Map<Channel, AtomicInteger> with strong references, cleaned only via Disconnect.cleanChannel. If a channel is torn down without going through cleanChannel (e.g. abnormal disconnect), its id-space entry leaks for the process lifetime. Consider a WeakHashMap keyed by Channel, or guaranteeing cleanup on all close paths.
Buffer sharing (pre-existing): send wraps the shared incoming payload with Unpooled.wrappedBuffer(payload) for every subscriber. Since WrappedByteBuf shares the underlying buffer's reference count, fan-out to multiple active subscribers releases the same buffer N times. This is pre-existing behavior, but worth double-checking the reference-count handling (e.g. payload.retain() per subscriber) so it doesn't over-release under multi-subscriber fan-out.

I also confirmed SubscribeRepository's BaseRepository type change (List<Channel> → Map<Channel, MqttQoS>) doesn't break other callers — the only callers (Subscribe.add(channel, subscriptions), Unsubscribe.remove(topics, channel), and Publish itself) use unchanged signatures.

Approving.

…n-out Use a weak-keyed map for per-channel packet id spaces so channels closed without a DISCONNECT do not leak entries, and retain the publish payload per subscriber so each outbound message owns a reference. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

) Respond to qos2 publishes with PUBREC instead of PUBACK, implement the PUBREL -> PUBCOMP half-handshake, and zero the reserved fixed-header bits of the PUBACK/PUBREC responses. Remove the unused qos0 stub. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Remove the closing channel from ChannelRepository, SubscribeRepository and the packet id space in MqttTransportHandler.operationComplete so ungraceful disconnects do not leak entries. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

wy471x · 2026-08-14T15:09:24Z

This is a solid fix that brings the MQTT fan-out into line with the spec: it now delivers at min(published QoS, granted QoS) and allocates a fresh packet identifier from each subscriber channel's own id space, instead of hard-coding AT_MOST_ONCE and reusing the publisher's packet id. The change is well-scoped and backed by thorough unit tests (PublishTest, SubscribeRepositoryTest, MqttPacketIdGeneratorTest) covering granted-QoS fan-out, QoS0 → packetId 0, per-subscriber id spaces, max-QoS merge, FAILURE filtering, and id wrap-around at 0xFFFF.

A few non-blocking notes:

Dead code: int packetId = msg.variableHeader().packetId(); in Publish.publish is now unused (it used to be passed to send). Please remove it, along with the now-resolved //// todo qos comment.

Id-space cleanup: MqttPacketIdGenerator.CHANNEL_PACKET_ID_FACTORY is a static Map<Channel, AtomicInteger> with strong references, cleaned only via Disconnect.cleanChannel. If a channel is torn down without going through cleanChannel (e.g. abnormal disconnect), its id-space entry leaks for the process lifetime. Consider a WeakHashMap keyed by Channel, or guaranteeing cleanup on all close paths.

Buffer sharing (pre-existing): send wraps the shared incoming payload with Unpooled.wrappedBuffer(payload) for every subscriber. Since WrappedByteBuf shares the underlying buffer's reference count, fan-out to multiple active subscribers releases the same buffer N times. This is pre-existing behavior, but worth double-checking the reference-count handling (e.g. payload.retain() per subscriber) so it doesn't over-release under multi-subscriber fan-out.

I also confirmed SubscribeRepository's BaseRepository type change (List<Channel> → Map<Channel, MqttQoS>) doesn't break other callers — the only callers (Subscribe.add(channel, subscriptions), Unsubscribe.remove(topics, channel), and Publish itself) use unchanged signatures.

Approving.

Thank you for the code review on this PR.

Fix: prevent mqtt packet id space leak and payload over-release on fan-out — addresses two review findings:

Packet id space leak (MqttPacketIdGenerator)

Replaced the static ConcurrentHashMap<Channel, AtomicInteger> with Collections.synchronizedMap(new WeakHashMap<>())
Id-space entries were previously only removed via Disconnect.cleanChannel; channels torn down without a DISCONNECT (abnormal disconnect) leaked their entry for the process
lifetime
Weak keys now let the entry die with the channel, while explicit remove() remains for clean disconnects

Payload over-release under fan-out (Publish.send)

Changed Unpooled.wrappedBuffer(payload) to Unpooled.wrappedBuffer(payload.retain())
The wrapped buffer shares the inbound payload's ref-count, so N subscribers releasing their outbound messages would release the same buffer N times
Each subscriber now owns one reference, keeping the ref-count balanced

Regression test (PublishTest)

Added testPublishFanOutRetainsPayloadPerSubscriber: publishes to 2 subscribers, asserts refCnt goes 1 → 3 after fan-out and back to 1 after the delivered messages are released

The "dead code" note from the review was verified against the code: the //// todo qos comment had already been removed in the prior commit, and packetId is still used by the
qos1/qos2 PUBACK path, so it was intentionally left intact.

Aias00

Summary (re-review after update)

Re-reviewed after the branch was updated (prior review was dismissed by a rebase). The current version already addresses both issues I raised before: MqttPacketIdGenerator now uses Collections.synchronizedMap(new WeakHashMap<>()) so a channel torn down without cleanChannel does not leak its id space, and send uses payload.retain() per subscriber so the shared payload survives fan-out without under/over-releasing.

What is correct:

Per-subscriber packet id space: MqttPacketIdGenerator keeps a WeakHashMap<Channel, AtomicInteger> (synchronized) so each channel gets its own id sequence, wrapping at 0xFFFF; remove(channel) is called on disconnect/close (via MqttTransportHandler.operationComplete and Disconnect.cleanChannel). This fixes the old behavior of reusing a shared/hard-coded id across subscribers.
QoS correctness: qos1 sends a proper PUBACK; qos2 now sends PUBREC (previously incorrectly sent PUBACK for QoS 2). New PubRel/PUBCOMP handler completes the QoS 2 handshake and is wired into MqttFactory (case PUBREL).
Delivery QoS = min(published, granted): send computes minQoS(publishQoS, grantedQoS) per subscriber and allocates a packet id only for QoS > 0. SubscribeRepository stores Map<Channel, MqttQoS> and merges overlapping subscriptions with maxQoS.
ByteBuf refcount fix: send uses payload.retain() per subscriber; testPublishFanOutRetainsPayloadPerSubscriber asserts the refcount — the old code under-counted on multi-subscriber fan-out.
Cleanup on close: MqttTransportHandler.operationComplete removes the channel from ChannelRepository, SubscribeRepository, and MqttPacketIdGenerator, preventing leaked state after disconnect.
Test coverage is comprehensive: PublishTest, PubRelTest, MqttTransportHandlerTest, SubscribeRepositoryTest, MqttPacketIdGeneratorTest.

Conclusion

Approved. The QoS 2 completion, per-subscriber id allocation, and ByteBuf refcount handling are all real correctness fixes, and the prior leak/refcount concerns are now resolved.

Non-blocking suggestions

Merge ordering (important): this PR rewrites SubscribeRepository to Map<Channel, MqttQoS> (changing get(topic)'s return type), while #6902 and #6906 keep List<Channel> and add getChannelsByTopic. These three MQTT PRs conflict on SubscribeRepository/Publish.java. Merge #6902/#6906 first (or rebase this one on top of them), then this PR, so the SubscribeRepository shape is decided once. After that, #6902's publishWill (random packet id) could be updated to use MqttPacketIdGenerator for consistency.
Async-payload note: Publish.publish still captures the incoming payload ByteBuf into a CompletableFuture.runAsync used after the method returns. The new retain() makes fan-out safe, but the caller must still not release the payload before the async send runs — same contract as before, just more robust. A short comment would help future maintainers.

Thanks for addressing the prior feedback.

# Conflicts: # shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandler.java # shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepository.java # shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandlerTest.java # shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java # shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepositoryTest.java

- SubscribeRepository.remove(List, Channel) kept two blocks after the merge, one of which assigns a Map to a List and does not compile - release the static repository maps around every test in PublishTest, SubscribeRepositoryTest and MqttTransportHandlerTest so the shared state of one test class cannot leak into another - cover the non-MQTT read path of MqttTransportHandler, the per-topic QoS merging of SubscribeRepository and the retained publish handling

Copilot

🟡 Changes recommended

QoS grants, acknowledgement state machines, packet-ID reuse, and asynchronous cleanup remain incorrect.

Get a fresh assessment by requesting another Copilot review.

Pull request overview

Updates MQTT delivery to preserve subscriber QoS, allocate channel-specific packet IDs, complete QoS 2 acknowledgements, and clean state on channel closure.

Changes:

Stores subscriber QoS and applies it during publish fan-out.
Adds packet-ID generation and PUBREC/PUBREL/PUBCOMP handling.
Adds repository cleanup and expanded unit tests.

File summaries

File	Description
`MqttPacketIdGeneratorTest.java`	Tests packet-ID allocation and reset.
`SubscribeRepositoryTest.java`	Tests QoS-aware subscriptions and cleanup.
`PubRelTest.java`	Tests PUBCOMP responses.
`PublishTest.java`	Tests fan-out QoS, packet IDs, and acknowledgements.
`MqttTransportHandlerTest.java`	Tests close-time repository cleanup.
`MqttPacketIdGenerator.java`	Adds per-channel packet-ID generation.
`SubscribeRepository.java`	Stores subscriber QoS by topic.
`PubRel.java`	Handles PUBREL with PUBCOMP.
`Publish.java`	Applies granted QoS and emits PUBACK/PUBREC.
`MqttTransportHandler.java`	Cleans repositories when channels close.
`MqttFactory.java`	Dispatches PUBREL messages.
`MessageType.java`	Delegates PUBREL processing.
`Disconnect.java`	Clears packet-ID state.
`AbstractMessageType.java`	Adds the PUBREL handler contract.

Review details

Files reviewed: 14/14 changed files
Comments generated: 8
Review effort level: Balanced

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

+            case PUBREL:
+                messageType.pubRel(ctx, msg);
+                break;


        }
        int packetId = msg.variableHeader().packetId();
-        CompletableFuture.runAsync(() -> send(topic, payload, packetId));
+        CompletableFuture.runAsync(() -> send(topic, payload, mqttQoS));


+        AtomicInteger packetId = CHANNEL_PACKET_ID_FACTORY.computeIfAbsent(channel, key -> new AtomicInteger());
+        return packetId.updateAndGet(current -> current >= MAX_PACKET_ID ? MIN_PACKET_ID : current + 1);


        Singleton.INST.get(ChannelRepository.class).remove(channel);
+        MqttPacketIdGenerator.remove(channel);


+    public void pubRel(final ChannelHandlerContext ctx, final MqttMessage msg) {
+        MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) msg.variableHeader();
+        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0);
+        MqttMessage mqttPubCompMessage = new MqttMessage(mqttFixedHeader, variableHeader);
+        ctx.writeAndFlush(mqttPubCompMessage);
+    }


+                .filter(s -> s.qualityOfService() != MqttQoS.FAILURE)
+                .forEach(s -> TOPIC_CHANNEL_FACTORY
+                        .computeIfAbsent(s.topicName(), key -> new ConcurrentHashMap<>())
+                        .merge(channel, s.qualityOfService(), SubscribeRepository::maxQoS)));


+                .forEach(s -> TOPIC_CHANNEL_FACTORY
+                        .computeIfAbsent(s.topicName(), key -> new ConcurrentHashMap<>())
+                        .merge(channel, s.qualityOfService(), SubscribeRepository::maxQoS)));


+    public void remove(final Channel channel) {
+        CompletableFuture.runAsync(() -> TOPIC_CHANNEL_FACTORY.values().parallelStream()
+                .forEach(subscribers -> subscribers.remove(channel)));


Aias00

Approved as PMC (Aias00). Coherent fix with regression tests; green CI, mergeable. Reviewed the diff.

wy471x and others added 2 commits August 13, 2026 23:50

Merge branch 'master' into fix_locate-packid-for-every-sub

0fb44cd

Aias00 previously approved these changes Aug 14, 2026

View reviewed changes

Aias00 and others added 3 commits August 14, 2026 14:49

Merge branch 'master' into fix_locate-packid-for-every-sub

499caf2

Merge branch 'master' into fix_locate-packid-for-every-sub

7496ba4

wy471x dismissed Aias00’s stale review via 9bcfb63 August 14, 2026 13:13

wy471x and others added 2 commits August 14, 2026 23:00

wy471x added 2 commits August 14, 2026 23:13

Merge branch 'master' into fix_locate-packid-for-every-sub

d906556

Merge branch 'master' into fix_locate-packid-for-every-sub

04e73d8

Aias00 mentioned this pull request Sep 4, 2026

feat: implement MQTT wildcard subscription matching #6906

Open

3 tasks

Aias00 previously approved these changes Sep 4, 2026

View reviewed changes

Aias00 mentioned this pull request Sep 4, 2026

feat: implement MQTT Last Will and Testament (LWT) #6902

Open

3 tasks

wy471x added 2 commits September 16, 2026 22:33

wy471x dismissed Aias00’s stale review via 53b77db September 17, 2026 13:24

Merge branch 'master' into fix_locate-packid-for-every-sub

6c2eeba

dengliming requested a balanced review from Copilot September 18, 2026 03:28

Copilot started reviewing on behalf of dengliming September 18, 2026 03:28 View session

Copilot AI reviewed Sep 18, 2026

View reviewed changes

Aias00 approved these changes Sep 19, 2026

View reviewed changes

Aias00 and others added 2 commits September 20, 2026 05:02

Merge branch 'master' into fix_locate-packid-for-every-sub

e4890a7

Merge branch 'master' into fix_locate-packid-for-every-sub

43a553f

Navigation Menu

fix: deliver mqtt messages at granted qos with per-subscriber packet id - #6913

fix: deliver mqtt messages at granted qos with per-subscriber packet id#6913
wy471x wants to merge 14 commits into
apache:masterfrom
wy471x:fix_locate-packid-for-every-sub

wy471x commented Aug 13, 2026 •

edited

Loading

Uh oh!

Aias00 left a comment

Uh oh!

wy471x commented Aug 14, 2026 •

edited

Loading

Uh oh!

Aias00 left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Aias00 left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		AtomicInteger packetId = CHANNEL_PACKET_ID_FACTORY.computeIfAbsent(channel, key -> new AtomicInteger());
		return packetId.updateAndGet(current -> current >= MAX_PACKET_ID ? MIN_PACKET_ID : current + 1);

		Singleton.INST.get(ChannelRepository.class).remove(channel);
		MqttPacketIdGenerator.remove(channel);

Conversation

wy471x commented Aug 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Uh oh!

Aias00 left a comment

Choose a reason for hiding this comment

Uh oh!

wy471x commented Aug 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Aias00 left a comment

Choose a reason for hiding this comment

Summary (re-review after update)

Conclusion

Non-blocking suggestions

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Aias00 left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

wy471x commented Aug 13, 2026 •

edited

Loading

wy471x commented Aug 14, 2026 •

edited

Loading