[konlue]

Fixes #7919.

Summary

An inbound QoS 2 PUBLISH was always fanned out to subscribers and only answered with PUBREC — no receiving-side state was kept. A client retransmitting the same packet id before its PUBREL (e.g. after a slow PUBREC) therefore got the message delivered to subscribers a second time under a freshly allocated packet id, breaking exactly-once.

Modifications

• New InflightRepository (BaseRepository, so MqttBootstrapServer registers it with the other repositories): per-client set of inbound packet ids currently in the acknowledgement flow.
• Publish now registers the packet id before delivery; only a first-seen packet id is fanned out, and every QoS 2 publish (first or retransmitted) is answered with PUBREC. QoS 0/1 paths are unchanged, and an unknown client (no CONNECT handshake seen) keeps the previous behavior.
• PubRel releases the packet id when PUBCOMP is sent, so the id becomes reusable; a PUBREL for an unknown packet id still gets PUBCOMP, per spec.
• Disconnect.cleanChannel discards the in-flight state together with the session on a clean-session disconnect; a persistent session keeps it across reconnects, matching the session-present resumption semantics.

Verifying this change

• testQos2RetransmitIsPublishedExactlyOnce: packet id 77 delivered once; the DUP retransmission is answered with PUBREC and the subscriber still sees exactly one publication. Fails on the previous code (two deliveries).

• testPubRelReleasesInflightForRedelivery: PUBREL → PUBCOMP, and the same packet id may be delivered again afterwards.

• testPubRelForUnknownPacketIdStillRespondsWithPubComp.

• Full module tests and Checkstyle pass:

  ./mvnw -pl shenyu-protocol/shenyu-protocol-mqtt checkstyle:check

  132/132 green, checkstyle clean.

[Aias00]

Thanks for the fix — the direction is right, and the placement of the state (per-client inbound packet ids, released on PUBREL) matches the spec's "packet id is free once PUBCOMP is sent" model. I verified a few things that are correct and found two issues I'd like addressed before merge.

Blocking

1. A cleanSession=1 reconnect keeps the in-flight state → silent message loss with a false PUBREC/PUBCOMP ack

Connect discards the stored session on cleanSession=1, but never discards the new in-flight state for that clientId:

if (cleanSession) {
    // A clean-session connect must discard any previously stored session.
    session = new MqttSession(clientId, true);
    sessionRepository.add(clientId, session);
} else if (Objects.isNull(session)) { ... }

Disconnect.cleanChannel only clears it when session.isCleanSession() is true, so a persistent-session client that disappears without a DISCONNECT (TCP drop, crash, keep-alive timeout) leaves its in-flight set behind:

if (Objects.nonNull(session) && session.isCleanSession()) {
    Singleton.INST.get(SessionRepository.class).remove(clientId);
    Singleton.INST.get(InflightRepository.class).remove(clientId);
}

Resulting sequence:

1. CONNECT cleanSession=0 → stored session.
2. PUBLISH QoS2 packetId=7 → INBOUND_INFLIGHT = {X: {7}}, delivered, PUBREC sent.
3. TCP drops, no DISCONNECT → isCleanSession() is false → {7} survives.
4. Reconnect with cleanSession=1 → MQTT-3.1.2-6 says both sides MUST discard the previous session and start a new one, so the packet-id space is reset; Connect does discard the session but not {7}.
5. Client publishes a brand-new message with packetId=7 (legal now).
6. registerInbound → register(X, 7) returns false → the message is never fanned out, while PUBREC and then PUBCOMP are still sent, so the publisher believes it was delivered exactly once.

That is the same class of bug the PR is fixing, only now on the receiving side and with a false positive ack instead of a duplicate. The InflightRepository Javadoc ("a clean-session disconnect discards them with the rest of the session state") states an invariant that the CONNECT path does not uphold.

Suggested fix — clear the in-flight state whenever the session is not resumed, which also covers the "no stored session at all" case and the leftover-key leak in issue 3:

boolean sessionPresent = !cleanSession && Objects.nonNull(session);
if (cleanSession) {
    session = new MqttSession(clientId, true);
    sessionRepository.add(clientId, session);
} else if (Objects.isNull(session)) {
    session = new MqttSession(clientId, false);
    sessionRepository.add(clientId, session);
}
if (!sessionPresent) {
    // a new session resets the packet id space, so stale inbound ids must not suppress new publishes
    Singleton.INST.get(InflightRepository.class).remove(clientId);
}

Please also add a regression test: register an in-flight id, reconnect with cleanSession=1, publish again with the same packet id, and assert the subscriber receives it.

2. testQos2RetransmitIsPublishedExactlyOnce does not actually prove the dedup

captureMessages verifies cumulatively over the whole test:

private List<MqttPublishMessage> captureMessages(final Channel channel, final int times) {
    verify(channel, timeout(TIMEOUT.toMillis()).times(times)).writeAndFlush(captor.capture());

so at the last assertion the count is already 1 from line 353, and timeout(5s).times(1) returns immediately. The retransmission is delivered on CompletableFuture.runAsync, i.e. after publish() has returned, so the assertion only fails when the second delivery happens to land before the verify — a race, not an ordering guarantee. On a build without the dedup the test is nondeterministic rather than red (the PR description's "fails on the previous code" holds only for the lucky interleavings).

A deterministic form uses a second packet id as an ordering barrier:

publishToSubscribers(MqttQoS.EXACTLY_ONCE, RETRANSMIT_PACKET_ID, true);
assertPubRec(RETRANSMIT_PACKET_ID);

// barrier: if the retransmission had been fanned out, this message would be the third delivery
publishToSubscribers(MqttQoS.EXACTLY_ONCE, ANOTHER_PACKET_ID, false);
assertPubRec(ANOTHER_PACKET_ID);
List<MqttPublishMessage> delivered = captureMessages(subscriberChannel, 2);

times(2) then fails on broken code without depending on timing.

Non-blocking

3. release() removes the packet id but never removes the now-empty set, so every clientId that ever published QoS2 and never sent a clean-session DISCONNECT keeps an empty Set for the lifetime of the JVM (same leak shape as #7939). Since the map is keyed by clientId and IoT/mobile ids rotate, consider:

   Set<Integer> packetIds = INBOUND_INFLIGHT.get(clientId);
   if (Objects.nonNull(packetIds) && packetIds.remove(packetId) && packetIds.isEmpty()) {
       INBOUND_INFLIGHT.remove(clientId, packetIds); // two-arg remove: a concurrent register must not be evicted
   }

4. testPublishAllocatesPacketIdFromSubscriberIdSpace publishes two QoS2 messages with the same PUBLISHER_PACKET_ID (12345) and expects both to be delivered. It only still passes because publisherChannel is not registered in ChannelRepository, so registerInbound takes the clientId == null escape hatch. The test therefore no longer models a conforming client and will break the moment the publisher is registered in setUp. Giving the two publishes distinct packet ids would remove the trap.

5. registerInbound silently degrades to "no dedup" when the channel has no clientId. That fallback is the right call, but it is the one path where the guarantee disappears without any trace — a LOG.debug there would make it observable in the field.

6. Dedup is keyed by clientId, not by channel. During a clientId takeover (see #7939) two channels briefly share one clientId, so a retransmission on the new channel is deduplicated against the old channel's state. Per spec the session state follows the clientId, so this is defensible, but it deserves a sentence in the InflightRepository Javadoc since it is the reason the key is the clientId and not the channel.

Verified as correct

• MqttBootstrapServer.initRepositories() registers every BaseRepository subtype found in ...protocol.mqtt.repositories via Reflections, so InflightRepository is wired in production automatically — Singleton.INST.get(InflightRepository.class) cannot be null at runtime, and ConnectTest/PubRelTest only need the explicit registration because they bypass init().
• Refcount handling in the deduplicated path: deliverAsync (and therefore payload.retain()) is skipped, and MqttTransportHandler.channelRead releases the inbound message in its finally, so the payload is freed at refCnt 0 — no leak in the new branch. The delivered path keeps the existing retain/release balance.
• registerInbound runs before qos2(ctx, packetId), so a retransmission is still answered with PUBREC, which is what MQTT-4.3.3 requires.
• release() before writing PUBCOMP is safe: the client cannot reuse the packet id until it sees PUBCOMP.
• ConcurrentHashMap.newKeySet() + computeIfAbsent(...).add() is the correct check-and-set primitive; Set.add is the atomic discriminator, so concurrent retransmissions on the same channel cannot both win.
• Scoping the dedup to QoS2 only is right — QoS1 is at-least-once by definition, so duplicates there are allowed and must not be suppressed.

Note on CI: build (shenyu-integrated-test-http) is currently red on master because of the CryptorResponsePluginTest.testWhenDoesntExistFiled assertion left behind by #7903 — that failure is unrelated to this PR.

[konlue]

All addressed:

1. In-flight ID cleanup on session reset: Connect now discards the in-flight ids whenever the session is not resumed — covering both the clean-session reconnect and the no-stored-session case — with the regression test you sketched (register an id, reconnect with cleanSession=1, republish the same id, assert delivery).
2. Deterministic retransmit test: the test now uses a second packet id as an ordering barrier before the cumulative times(2) assertion, so it fails deterministically without the dedup.

Also took the non-blocking points:

• The empty-set two-arg cleanup in release.
• A debug log on the no-clientId fallback.
• Distinct packet ids in the packet-id-allocation test.
• A Javadoc note on why the key is the client id.

[Aias00]

Review: #7938 fix(mqtt): deduplicate retransmitted qos2 publishes (re-review)

Verdict: approved — cleanly prevents message drop on reconnect, guarantees deterministic testing, and hardens QoS2 deduplication lifecycle.

What was verified in this update

1. Clean session packet id reset:
   • In Connect.connect, stale in-flight packet IDs are now evicted whenever a new session starts:

     if (!sessionPresent) {
         Singleton.INST.get(InflightRepository.class).remove(clientId);
     }

     This resolves the false ACK/silent drop defect when a client disconnects without DISCONNECT and later reconnects with cleanSession = true.
2. Deterministic ordering barrier in tests:
   • testQos2RetransmitIsPublishedExactlyOnce introduces BARRIER_PACKET_ID following the DUP transmission, ensuring times(2) fails reliably on broken implementations without depending on async scheduler timing.
   • Added testCleanSessionReconnectClearsInflightIds to verify that packet ID reuse across separate clean sessions succeeds.
3. Memory leak and fallback cleanup:
   • In InflightRepository.release, empty sets are removed via two-argument conditional removal (INBOUND_INFLIGHT.remove(clientId, packetIds)).
   • In Publish.registerInbound, missing client ID scenarios emit debug logging before gracefully falling back.

[Aias00]

Follow-up: pr_build is red on this PR, and my approval was premature

I approved this before the latest CI run finished. That was my mistake — the new head (6d5de3a1) fails pr_build, and the failure comes from this PR's own change, not from the master regression. Withdrawing the approval until it is green.

The failure

[ERROR] Tests run: 8, Failures: 0, Errors: 1, Skipped: 0
        <<< FAILURE! - in org.apache.shenyu.protocol.mqtt.MqttFactoryTest
[ERROR] org.apache.shenyu.protocol.mqtt.MqttFactoryTest
        .connectShouldBeDispatchedToConnectHandler  Time elapsed: 0.017 s  <<< ERROR!
[ERROR]   MqttFactoryTest.connectShouldBeDispatchedToConnectHandler:110 » NullPointer Ca...
[ERROR] Failed to execute goal maven-surefire-plugin:3.0.0-M4:test
        on project shenyu-protocol-mqtt: There are test failures.
[INFO] shenyu-protocol-mqtt ............................... FAILURE [ 38.078 s]
[INFO] BUILD FAILURE

ci/build fails on the same run.

Why

Connect.connect now reaches InflightRepository on every new session:

if (!sessionPresent) {
    Singleton.INST.get(InflightRepository.class).remove(clientId);
}

Singleton.INST.get(...) is a plain SINGLES.get(clazz.getName()) — it returns null for an unregistered type, it does not create one:

public <T> T get(final Class<T> clazz) {
    return (T) SINGLES.get(clazz.getName());
}

MqttFactoryTest.setUp registers ChannelRepository, WillRepository, SessionRepository and SubscribeRepository but not InflightRepository:

@BeforeEach
public void setUp() {
    channelRepository = new ChannelRepository();
    willRepository = new WillRepository();
    Singleton.INST.single(ChannelRepository.class, channelRepository);
    Singleton.INST.single(WillRepository.class, willRepository);
    Singleton.INST.single(SessionRepository.class, new SessionRepository());
    Singleton.INST.single(SubscribeRepository.class, new SubscribeRepository());
    ...
}

So connectMessage() (clean session, no stored session → sessionPresent == false) dereferences null and throws. ConnectTest, ConnectSessionPresentTest and PubRelTest were all updated to register the repository in this PR; MqttFactoryTest was missed.

At runtime this is not a defect — MqttBootstrapServer.initRepositories() registers every BaseRepository subtype found by Reflections, so InflightRepository is always present in a real deployment. This is a test-fixture gap, not a code bug.

Fix

One line in MqttFactoryTest.setUp, matching the other three tests:

Singleton.INST.single(InflightRepository.class, new InflightRepository());

Worth grepping the module for other tests that drive a real Connect and asserting each one registers the repository, so this cannot recur the next time Connect reaches a new singleton.

Everything else still stands

My re-review verdict on the code itself is unchanged — the cleanSession reset, the BARRIER_PACKET_ID ordering barrier in testQos2RetransmitIsPublishedExactlyOnce, the two-arg INBOUND_INFLIGHT.remove(...) on release, and the LOG.debug on the no-clientId fallback are all right, and testCleanSessionReconnectClearsInflightIds covers exactly the defect I raised. Only the fixture is missing.

Please push the setUp fix (or an empty commit if you also want to re-run the flaky ci/build), and I will approve as soon as pr_build is green.

[konlue]

Pushed the fixture fix, and you were right to catch it — thanks also for the pointer to audit the rest.

• Fixture fix: MqttFactoryTest.setUp now registers InflightRepository. I grepped the module for tests that drive a real Connect and checked all five.
• Rebase discovery: After rebasing onto latest master, it also surfaced ClientIdTakeoverTest (merged through fix(mqtt): take over duplicate clientIds and fence stale-connection cleanup (#7918) #7939), which reached Connect without the registration — same trap, now fixed too.
• Merge conflict: The merge brought a conflict in Disconnect.cleanChannel, resolved by keeping [Bug] MQTT duplicate clientId connections remain active and old-channel cleanup deletes the new session #7918's ownership fence and moving [Bug] Retransmitted QoS2 PUBLISH packets are delivered twice with new downstream packet IDs #7919's in-flight cleanup inside it.
• Checkstyle fix: Connect.connect crossed the checkstyle MethodLength limit once the takeover block landed, so the takeover was extracted into takeOverClientId() (pure move).

Full module 137/137 green and checkstyle clean locally — pr_build should go green on this head.

[Aias00]

Re-review: #7938 fix(mqtt): deduplicate retransmitted qos2 publishes

Verdict: approved — the CI regression I flagged is fixed exactly as described, and the code change itself was already verified in my previous review.

The blocker is closed

MqttFactoryTest.setUp now registers the repository, with a comment that records why rather than just what:

Singleton.INST.single(SessionRepository.class, new SessionRepository());
Singleton.INST.single(SubscribeRepository.class, new SubscribeRepository());
// Connect reaches this repository on a new session, and Singleton.get does not create one
Singleton.INST.single(InflightRepository.class, new InflightRepository());

The comment is the part I care about most. Singleton.INST.get(...) is a bare SINGLES.get(name) that returns null for an unregistered type, so the next person who adds another Singleton.INST.get(...) call inside Connect will hit the same failure; naming the mechanism in the fixture is what stops that from being rediscovered. InflightRepository is imported, so the file compiles.

I also checked the other test classes in the module for the same gap: ClientIdTakeoverTest and DisconnectDispatchTest already reference InflightRepository, and ConnectTest / ConnectSessionPresentTest / PubRelTest were updated earlier in this PR. DisconnectTest, MqttSessionDisconnectTest, PublishWillTest and SubscribeQosTest do not reference it, but none of them reaches the new !sessionPresent branch — the previous run confirmed that empirically, with 133 mqtt tests and exactly one error, on MqttFactoryTest.

Still holds from my earlier review

• Connect.connect clears INBOUND_INFLIGHT for the client id whenever the session is not resumed, so a cleanSession = true reconnect no longer suppresses a legal reuse of an old packet id (the false-PUBREC / silent-drop defect).
• testQos2RetransmitIsPublishedExactlyOnce uses BARRIER_PACKET_ID as an ordering barrier, so times(2) fails deterministically on broken code instead of depending on where an async fan-out lands.
• InflightRepository.release drops the per-client entry once its last id is released, using the two-argument remove(clientId, packetIds) so a concurrent register cannot be evicted.
• Publish.registerInbound logs at debug when the channel has no client id, making the one path where dedup is unavailable observable.

Note

pr_build is still queued on this head (14 checks queued, runner backlog). The change is a single fixture line against a compile-checked import, and the module's unit tests are what previously caught the problem, so I am approving on inspection — but pr_build and build (shenyu-integrated-test-k8s-ingress-*) must go green before this is merged, since they are the jobs that actually exercised the NPE.