feat: implement MQTT Last Will and Testament (LWT) by wy471x · Pull Request #6902 · apache/shenyu

wy471x · 2026-08-10T15:11:57Z

Parse and store will fields (topic, message, QoS, retain) in WillRepository on CONNECT. Publish will message to subscribers on ungraceful disconnect via channelInactive hook. Clear will on graceful DISCONNECT. Fix DISCONNECT message dispatch in MqttFactory that was previously dropped.

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

Core changes:

WillRepository.java (new) — A ConcurrentHashMap-backed repository that stores WillEntry (topic, message bytes, QoS, retain flag) keyed by Netty Channel. Implements the existing
BaseRepository interface.
Connect.java — On CONNECT, if isWillFlag() is true, extracts the will fields from the MQTT CONNECT payload and stores them via WillRepository.
Disconnect.java — On graceful DISCONNECT, removes the will from WillRepository (so it won't fire). Replaces the old // todo Last words placeholder.
MqttTransportHandler.java — Overrides channelInactive() to detect ungraceful disconnects. If a will is still present for that channel when it goes inactive, it publishes the
will via Publish.publishWill() and then clears it.
Publish.java — New publishWill() static method that looks up subscribers for the will topic and writes the will message as an MQTT PUBLISH packet to each active subscriber
channel.
MqttFactory.java — Fixes a bug where DISCONNECT was falling through to PUBACK/default instead of calling messageType.disconnect(ctx).
pom.xml — Adds JUnit Jupiter, Mockito, and maven-surefire-plugin with --add-opens JVM args for testing.

Tests (5 new test files):

ConnectTest, DisconnectTest, MqttTransportHandlerTest, PublishWillTest, and WillRepositoryTest — covering will storage on connect, will clearance on disconnect, will firing on channel inactivity, inactive channel skipping, empty subscriber handling, and QoS/retain flag propagation.

close #6852

Parse and store will fields (topic, message, QoS, retain) in WillRepository on CONNECT. Publish will message to subscribers on ungraceful disconnect via channelInactive hook. Clear will on graceful DISCONNECT. Fix DISCONNECT message dispatch in MqttFactory that was previously dropped. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Copilot

Pull request overview

Implements MQTT Last Will and Testament (LWT) support in the shenyu-protocol-mqtt module by persisting will data on CONNECT, clearing it on graceful DISCONNECT, and publishing it on ungraceful disconnect via Netty channelInactive.

Changes:

Add WillRepository to store per-connection LWT entries.
Publish stored wills on channelInactive and clear wills on DISCONNECT.
Add unit tests and module test dependencies/config to validate LWT behavior.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java	Store will fields from CONNECT into `WillRepository`.
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java	Clear will on graceful DISCONNECT before closing channel.
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandler.java	Publish + clear will on ungraceful disconnect (`channelInactive`).
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java	Add `publishWill(...)` helper to emit will PUBLISH packets to subscribers.
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttFactory.java	Ensure DISCONNECT is dispatched to `messageType.disconnect(ctx)`.
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepository.java	New repository for storing will topic/message/QoS/retain by Channel.
shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java	Tests will persistence on CONNECT.
shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectTest.java	Tests will clearance on DISCONNECT.
shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandlerTest.java	Tests will firing/removal on `channelInactive`.
shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishWillTest.java	Tests will publishing behavior to subscribers.
shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepositoryTest.java	Tests repository CRUD semantics for will entries.
shenyu-protocol/shenyu-protocol-mqtt/pom.xml	Add JUnit/Mockito deps and configure Surefire argLine for tests.

Suppressed comments (1)

shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepository.java:75

WillEntry#getMessage currently returns the internal byte[] directly, allowing external callers to mutate repository state. Return a defensive copy instead.

        public byte[] getMessage() {
            return message;
        }

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

+        // store will if present
+        if (msg.variableHeader().isWillFlag()) {
+            WillRepository.WillEntry will = new WillRepository.WillEntry(
+                    msg.payload().willTopic(),
+                    msg.payload().willMessageInBytes(),
+                    msg.variableHeader().willQos(),
+                    msg.variableHeader().isWillRetain());
+            Singleton.INST.get(WillRepository.class).add(ctx.channel(), will);
+        }


+    @Override
+    public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
+        WillRepository.WillEntry will = Singleton.INST.get(WillRepository.class).get(ctx.channel());
+        if (Objects.nonNull(will)) {
+            Publish.publishWill(will);
+            Singleton.INST.get(WillRepository.class).remove(ctx.channel());
+        }
+        super.channelInactive(ctx);
+    }


+        public WillEntry(final String topic, final byte[] message, final int qos, final boolean retain) {
+            this.topic = topic;
+            this.message = message;
+            this.qos = qos;
+            this.retain = retain;
+        }


+    @AfterEach
+    public void tearDown() {
+        channel.close();
+    }


Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

…emented

Aias00

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

Verdict: ✅ Approve (with one consistency follow-up)

Solid, well-tested feature implementation. The lifecycle handling is correct and the test coverage is excellent.

What's correct

The MqttFactory DISCONNECT fix is the linchpin. Original code had case PUBACK: case DISCONNECT: default: break; — so DISCONNECT messages were silently dropped and Disconnect.disconnect() was never invoked. Without this fix, graceful-disconnect will removal couldn't work at all. Good catch, and it's required for the rest of the feature to function.
Correct will lifecycle:
- CONNECT with isWillFlag() → WillRepository.add(channel, will) ✅
- Graceful DISCONNECT → WillRepository.remove(channel) → will never fires ✅
- Ungraceful disconnect → MqttTransportHandler.channelInactive() sees the will present → Publish.publishWill(will) → removes it ✅
- The inactive channel is excluded from receiving its own will (channel.isActive() guard) ✅
- No double-publish: channelInactive removes the will immediately and Netty fires it once per close ✅
WillRepository is a clean ConcurrentHashMap<Channel, WillEntry> keyed by Channel (not clientId), so reconnects with a new channel don't collide, and testReplaceWillEntryOnReconnect covers the replace path.
publishWill null-guards topic/message, derives a valid packetId (0 for AT_MOST_ONCE, random otherwise), and respects the will QoS/retain flags.
Test coverage is genuinely thorough: ConnectTest (store / no-will / qos0 / retain), DisconnectTest (clears will / no will / removes channel), MqttTransportHandlerTest (fires+removes / no will / post-disconnect), PublishWillTest (active / inactive-skip / empty / qos+retain), and WillRepositoryTest. The pom changes (junit-jupiter, mockito, --add-opens for JDK 17) are the right scaffolding to support them.

Suggestion (non-blocking)

Wildcard-aware will delivery. Publish.publishWill uses SubscribeRepository.get(will.getTopic()) — an exact lookup. A client subscribed to e.g. status/# will not receive a will published to status/client-001, even though normal publish routing should match it. Since #6906 (wildcard matching) adds getChannelsByTopic, it would be consistent to route the will through that here too. Not blocking (LWT-to-wildcard-subscriber is an edge case), but worth aligning.
Retained will semantics. A will with retain=true is sent with the RETAIN flag, but there's no evidence the broker persists retained messages for later subscribers. That's a broader retained-message gap, outside this PR's scope — just flagging so it's tracked.

Verdict

Approving. The implementation is correct, the essential DISCONNECT routing bug is fixed, and the tests back the behavior end-to-end. Address the wildcard-delivery point as a small follow-up.

…emented

Publish.publishWill used an exact SubscribeRepository.get() lookup, so clients subscribed to wildcard filters (e.g. status/#) never received wills published to concrete topics like status/client-001. Port the TopicMatcher and SubscribeRepository.getChannelsByTopic from apache#6906 and route will delivery through it, consistent with normal publish routing. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

wy471x · 2026-08-15T16:15:39Z

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

Verdict: ✅ Approve (with one consistency follow-up)

Solid, well-tested feature implementation. The lifecycle handling is correct and the test coverage is excellent.

What's correct

The MqttFactory DISCONNECT fix is the linchpin. Original code had case PUBACK: case DISCONNECT: default: break; — so DISCONNECT messages were silently dropped and Disconnect.disconnect() was never invoked. Without this fix, graceful-disconnect will removal couldn't work at all. Good catch, and it's required for the rest of the feature to function.

Correct will lifecycle:

CONNECT with isWillFlag() → WillRepository.add(channel, will) ✅

Graceful DISCONNECT → WillRepository.remove(channel) → will never fires ✅

Ungraceful disconnect → MqttTransportHandler.channelInactive() sees the will present → Publish.publishWill(will) → removes it ✅

The inactive channel is excluded from receiving its own will (channel.isActive() guard) ✅

No double-publish: channelInactive removes the will immediately and Netty fires it once per close ✅

WillRepository is a clean ConcurrentHashMap<Channel, WillEntry> keyed by Channel (not clientId), so reconnects with a new channel don't collide, and testReplaceWillEntryOnReconnect covers the replace path.

publishWill null-guards topic/message, derives a valid packetId (0 for AT_MOST_ONCE, random otherwise), and respects the will QoS/retain flags.

Test coverage is genuinely thorough: ConnectTest (store / no-will / qos0 / retain), DisconnectTest (clears will / no will / removes channel), MqttTransportHandlerTest (fires+removes / no will / post-disconnect), PublishWillTest (active / inactive-skip / empty / qos+retain), and WillRepositoryTest. The pom changes (junit-jupiter, mockito, --add-opens for JDK 17) are the right scaffolding to support them.

Suggestion (non-blocking)

Wildcard-aware will delivery. Publish.publishWill uses SubscribeRepository.get(will.getTopic()) — an exact lookup. A client subscribed to e.g. status/# will not receive a will published to status/client-001, even though normal publish routing should match it. Since feat: implement MQTT wildcard subscription matching #6906 (wildcard matching) adds getChannelsByTopic, it would be consistent to route the will through that here too. Not blocking (LWT-to-wildcard-subscriber is an edge case), but worth aligning.

Retained will semantics. A will with retain=true is sent with the RETAIN flag, but there's no evidence the broker persists retained messages for later subscribers. That's a broader retained-message gap, outside this PR's scope — just flagging so it's tracked.

Verdict

Approving. The implementation is correct, the essential DISCONNECT routing bug is fixed, and the tests back the behavior end-to-end. Address the wildcard-delivery point as a small follow-up.

Thank you for the code review on this PR.

Fix: Wildcard-aware will delivery — previously Publish.publishWill used SubscribeRepository.get(topic), an exact-match lookup, so a client subscribed to status/# never received a will published to status/client-001.

Ported TopicMatcher and SubscribeRepository.getChannelsByTopic (identical to PR feat: implement MQTT wildcard subscription matching #6906 code) into the LWT branch
publishWill now routes through getChannelsByTopic, consistent with normal publish routing
Updated PublishWillTest/MqttTransportHandlerTest stubs; added wildcard-subscriber test and ported TopicMatcherTest/SubscribeRepositoryTest
Fixed checkstyle violation (== null → Objects.isNull)

…emented

Aias00

Summary (re-review after update)

Re-reviewed after the branch was updated (prior review was dismissed by a rebase). The current version is an improvement over what I last saw: Publish.publishWill now routes the will through getChannelsByTopic, so wildcard subscribers (e.g. status/#) correctly receive a will published to status/client-001 — the exact edge case I previously flagged is now handled.

What is correct:

Critical fix in MqttFactory: DISCONNECT was previously grouped with PUBACK/default and fell through to break, so messageType.disconnect(ctx) was never called — graceful disconnects were silently dropped. The PR gives DISCONNECT its own case that dispatches correctly. This is the linchpin that makes the rest of the feature work.
Connect stores the will only when isWillFlag() is true (topic, raw message bytes, QoS, retain).
Disconnect.disconnect removes the will before closing, so a clean shutdown does not fire the will.
MqttTransportHandler.channelInactive publishes the will (via Publish.publishWill) and removes it; the inactive channel is excluded from receiving its own will. No double-publish.
WillRepository is a clean ConcurrentHashMap<Channel, WillEntry> (keyed by Channel, so reconnects with a new channel don't collide).
publishWill null-guards topic/message, resolves targets through getChannelsByTopic (wildcard-aware), and uses a per-publish id (0 for QoS 0, random otherwise). Unpooled.wrappedBuffer(byte[]) is safe here (backing byte[] owned by WillEntry).
Test coverage is thorough: ConnectTest, DisconnectTest, MqttTransportHandlerTest, PublishWillTest, WillRepositoryTest, plus shared TopicMatcherTest/SubscribeRepositoryTest.

Conclusion

Approved. The DISCONNECT dispatch fix is a genuine correctness bug fix and the LWT feature is implemented cleanly with good tests; the wildcard-delivery gap from my prior review is now closed.

Non-blocking suggestions

Merge ordering: this PR and #6906 add the same getChannelsByTopic to SubscribeRepository (both keep get(topic) returning List<Channel>), while #6913 rewrites SubscribeRepository to Map<Channel, MqttQoS>. Merge #6902/#6906 first and rebase #6913 (or merge #6913 last) to avoid a conflict. Coordinate the three MQTT PRs (#6902, #6906, #6913).
Spec nuance (optional): the will is armed in Connect before the CONNACK is sent. Today Connect always returns CONNECTION_ACCEPTED, so this is fine, but if a rejected-CONNECT path is ever added the will should only be armed after a successful CONNACK to avoid firing on auth failure. Not a defect today.

Thanks for addressing the prior feedback.

…emented # 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/ConnectTest.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/repositories/SubscribeRepositoryTest.java

Aias00

REQUEST_CHANGES (PMC Aias00): this PR has merge conflicts (CONFLICTING) and cannot be merged. Please rebase onto the latest master and resolve the conflicts, then request a re-review.

- allocate the per-topic channel list with computeIfAbsent and apply add/remove on the calling thread, so concurrent subscribers of the same new topic no longer overwrite each other's list and a subscription is visible as soon as add returns; get(List) no longer throws when a topic has no subscribers; drop the unused logger - SubscribeRepositoryTest: replace the Mockito mock and the awaitility/common-pool polling with deterministic assertions, and add a concurrency regression test for eight subscribers of the same new topic - ConnectTest: drive Connect with real MQTT messages instead of mocks, keep the credentials fixed for the whole class, cover the identifier-rejected and bad-credentials branches and assert a rejected CONNECT stores no will - MqttFactoryTest: DISCONNECT is dispatched to Disconnect, so assert the channel is closed and the channel/will repositories are cleared, and register the WillRepository that Disconnect looks up via Singleton - MqttTransportHandler: drop the will before publishing it and notify the pipeline exactly once on channelInactive

…emented # Conflicts: # shenyu-protocol/shenyu-protocol-mqtt/pom.xml

dengliming requested a lite review from Copilot August 11, 2026 04:10

Copilot started reviewing on behalf of dengliming August 11, 2026 04:10 View session

Copilot AI reviewed Aug 11, 2026

View reviewed changes

yu199195 and others added 2 commits August 11, 2026 18:22

Potential fix for pull request finding

ab95921

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Merge branch 'master' into feat_Last-Will-&-Testament-entirely-unimpl…

22a7e8f

…emented

Aias00 previously approved these changes Aug 14, 2026

View reviewed changes

Aias00 and others added 5 commits August 14, 2026 14:54

Merge branch 'master' into feat_Last-Will-&-Testament-entirely-unimpl…

74d51d4

…emented

Merge branch 'master' into feat_Last-Will-&-Testament-entirely-unimpl…

b37ff8b

…emented

Merge branch 'master' into feat_Last-Will-&-Testament-entirely-unimpl…

20e61b2

…emented

Merge branch 'master' into feat_Last-Will-&-Testament-entirely-unimpl…

cfc9cb8

…emented

wy471x dismissed Aias00’s stale review via 61384ab August 15, 2026 15:58

Merge branch 'master' into feat_Last-Will-&-Testament-entirely-unimpl…

21b42ba

…emented

This was referenced Sep 4, 2026

feat: implement MQTT wildcard subscription matching #6906

Open

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

Open

Aias00 approved these changes Sep 4, 2026

View reviewed changes

Aias00 requested changes Sep 19, 2026

View reviewed changes

wy471x added 2 commits September 19, 2026 20:24

Merge branch 'master' into feat_Last-Will-&-Testament-entirely-unimpl…

1510139

…emented # Conflicts: # shenyu-protocol/shenyu-protocol-mqtt/pom.xml

Navigation Menu

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

feat: implement MQTT Last Will and Testament (LWT)#6902
wy471x wants to merge 12 commits into
apache:masterfrom
wy471x:feat_Last-Will-&-Testament-entirely-unimplemented

wy471x commented Aug 10, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Aias00 left a comment

Uh oh!

wy471x commented Aug 15, 2026

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

What's correct

Suggestion (non-blocking)

Verdict

Uh oh!

Aias00 left a comment

Uh oh!

Aias00 left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Conversation

wy471x commented Aug 10, 2026

Summary

Core changes:

Tests (5 new test files):

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Aias00 left a comment

Choose a reason for hiding this comment

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

What's correct

Suggestion (non-blocking)

Verdict

Uh oh!

wy471x commented Aug 15, 2026

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

What's correct

Suggestion (non-blocking)

Verdict

Uh oh!

Aias00 left a comment

Choose a reason for hiding this comment

Summary (re-review after update)

Conclusion

Non-blocking suggestions

Uh oh!

Aias00 left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants