Conversation
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>
There was a problem hiding this comment.
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
WillRepositoryto store per-connection LWT entries. - Publish stored wills on
channelInactiveand 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>
Aias00
left a comment
There was a problem hiding this comment.
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
MqttFactoryDISCONNECT fix is the linchpin. Original code hadcase PUBACK: case DISCONNECT: default: break;— so DISCONNECT messages were silently dropped andDisconnect.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:
channelInactiveremoves the will immediately and Netty fires it once per close ✅
- CONNECT with
WillRepositoryis a cleanConcurrentHashMap<Channel, WillEntry>keyed by Channel (not clientId), so reconnects with a new channel don't collide, andtestReplaceWillEntryOnReconnectcovers the replace path.publishWillnull-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), andWillRepositoryTest. The pom changes (junit-jupiter, mockito,--add-opensfor JDK 17) are the right scaffolding to support them.
Suggestion (non-blocking)
- Wildcard-aware will delivery.
Publish.publishWillusesSubscribeRepository.get(will.getTopic())— an exact lookup. A client subscribed to e.g.status/#will not receive a will published tostatus/client-001, even though normal publish routing should match it. Since #6906 (wildcard matching) addsgetChannelsByTopic, 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=trueis 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.
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>
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.
|
Aias00
left a comment
There was a problem hiding this comment.
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 withPUBACK/defaultand fell through tobreak, somessageType.disconnect(ctx)was never called — graceful disconnects were silently dropped. The PR givesDISCONNECTits owncasethat dispatches correctly. This is the linchpin that makes the rest of the feature work. Connectstores the will only whenisWillFlag()is true (topic, raw message bytes, QoS, retain).Disconnect.disconnectremoves the will before closing, so a clean shutdown does not fire the will.MqttTransportHandler.channelInactivepublishes the will (viaPublish.publishWill) and removes it; the inactive channel is excluded from receiving its own will. No double-publish.WillRepositoryis a cleanConcurrentHashMap<Channel, WillEntry>(keyed by Channel, so reconnects with a new channel don't collide).publishWillnull-guards topic/message, resolves targets throughgetChannelsByTopic(wildcard-aware), and uses a per-publish id (0 for QoS 0, random otherwise).Unpooled.wrappedBuffer(byte[])is safe here (backingbyte[]owned byWillEntry).- Test coverage is thorough:
ConnectTest,DisconnectTest,MqttTransportHandlerTest,PublishWillTest,WillRepositoryTest, plus sharedTopicMatcherTest/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
getChannelsByTopictoSubscribeRepository(both keepget(topic)returningList<Channel>), while #6913 rewritesSubscribeRepositorytoMap<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
Connectbefore the CONNACK is sent. TodayConnectalways returnsCONNECTION_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
left a comment
There was a problem hiding this comment.
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
|
Re-check: this branch is still CONFLICTING with master (GitHub reports mergeStateStatus=DIRTY), so it cannot be merged regardless of CI. Please rebase onto the latest master and resolve the conflicts. |
…emented # Conflicts: # shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java # shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttFactory.java # 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/Publish.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/repositories/SubscribeRepositoryTest.java
- adapt getChannelsByTopic to the QoS-aware Map<Channel, MqttQoS> storage - merge both sides of MqttTransportHandlerTest into a single lifecycle - merge SubscribeRepositoryTest onto the async QoS API and drop duplicates - remove duplicated imports and unused imports left by the merge
sunnysabor
left a comment
There was a problem hiding this comment.
I rechecked the new conflict-resolution head b29821d3 and ran ./mvnw -B -ntp -pl shenyu-protocol/shenyu-protocol-mqtt -am -DskipTests=false -Dcheckstyle.skip=false test; the reactor passed, including all 103 MQTT module tests and Checkstyle. I found one remaining delivery correctness issue in publishWill:
It gets only List<Channel> from getChannelsByTopic and writes every will using willQos (lines 160-171). Normal Publish.send instead uses minQoS(publishQoS, subscriptionGrantedQoS) for each subscriber (lines 131-148). So a QoS 2 will is sent at QoS 2 even to a subscriber whose matching subscription was granted QoS 0, violating the subscription QoS limit. The new getChannelsByTopic API also discards the per-channel QoS needed to apply that limit.
Please preserve the maximum granted QoS per channel across matching (including wildcard and overlapping) filters, cap the will delivery QoS with it as the normal publish path does, and add a regression test where a QoS 2 will reaches a QoS 0 subscriber at QoS 0. The existing will tests assert only the will's QoS and do not cover a lower granted subscriber QoS.
getChannelsByTopic now returns matching channels mapped to the maximum granted qos across exact, wildcard and overlapping filters, and publishWill delivers each will with min(will qos, granted qos) like the normal publish path, so a qos 2 will is no longer sent at qos 2 to a qos 0 subscriber.
|
Thanks for the detailed re-check, @sunnysabor — the QoS-capping issue is fixed in 71fb63a.
Verified locally with The branch is also rebased onto the latest master and GitHub now reports it as mergeable, so the earlier conflict block is cleared. One heads-up: |
sunnysabor
left a comment
There was a problem hiding this comment.
Re-reviewed the updated head 71fb63a. The QoS correction preserves the maximum granted QoS per client across exact, wildcard, and overlapping filters, then caps each will delivery with min(will QoS, granted QoS); the new regression test covers QoS 2 will delivery to QoS 0 and QoS 2 subscribers and packet-id behavior. Verified locally with ./mvnw -B -ntp -pl shenyu-protocol/shenyu-protocol-mqtt -Dcheckstyle.skip=false test: 104 tests passed, Checkstyle clean. GitHub reports the branch mergeable and current triggered checks are green.
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:
./mvnw clean install -Dmaven.javadoc.skip=true.Summary
Core changes:
BaseRepository interface.
will via Publish.publishWill() and then clears it.
channel.
Tests (5 new test files):
close #6852