[konlue]

Fixes #7918.

Summary

Two connections with the same clientId could coexist: Connect only mapped channel→clientId, so a second CONNECT with the same id was accepted while the first channel stayed open. Worse, when the old channel later disconnected, cleanChannel resolved the clientId and unconditionally discarded the session — deleting the live session of the connection that replaced it.

Modifications

• ChannelRepository now keeps a clientId→channel ownership index. add registers ownership; takeover atomically transfers it and returns the previously owning channel.
• Connect takes over the id: when a different channel already owns it, its stored will is discarded (the client is alive on the new connection) and the previous connection is closed, per MQTT-3.1.4.
• cleanChannel fences the session-scoped cleanup on ownership: a replaced connection closing late can no longer discard the session of its replacement. Channel-scoped cleanup (subscriptions, channel mapping, outbound packet ids) is unchanged — it only touches the dead channel's own entries.

Verifying this change

• testTakeoverClosesThePreviousChannel: the second connect is accepted and the first channel is closed. Fails on the previous code (both channels stayed open).

• testStaleChannelCleanupKeepsTheNewSession (the issue's repro): after the takeover, running the stale channel's cleanup leaves the session intact. Fails on the previous code (session deleted).

• testWillOfReplacedConnectionIsDiscarded: the will of the replaced connection is dropped and never fires.

• Full module tests and Checkstyle pass:

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

  132/132 green (Connect / Disconnect / session-present regressions included); checkstyle clean.

[Aias00]

Review: take over duplicate clientIds and fence stale-connection cleanup

The core fix is right and well reasoned. Two things I checked specifically:

• Ordering is correct. takeover publishes the new owner into CLIENT_FACTORY before previous.close() runs, so when the replaced channel's channelInactive fires and cleanChannel(previous) executes, isOwner(clientId, previous) is already false and the new session survives. Reversing those two statements would reintroduce the exact bug in #7918, so it is worth a comment right there.
• The fence is on the right state. Fencing only the session-scoped cleanup while leaving SubscribeRepository.remove(channel), channelRepository.remove(channel) and MqttPacketIdGenerator.remove(channel) unfenced is correct — those are keyed by the dead channel itself.

But the new ownership index leaks.

remove() does not clean CLIENT_FACTORY

@Override
public void remove(final Channel channel) {
    CHANNEL_FACTORY.remove(channel);
}

CHANNEL_FACTORY is cleaned on every disconnect; CLIENT_FACTORY never is. Consequences:

1. Unbounded growth. One entry per distinct clientId ever seen, for the lifetime of the JVM. clientIds are client-chosen and routinely rotate (reconnects with a fresh id, IoT devices re-provisioned, mobile clients), so this grows without bound on a long-running gateway. CHANNEL_FACTORY does not have this problem because it is keyed by the channel and every channel is removed on close.
2. Every reconnect after a clean disconnect logs a false takeover. After a normal DISCONNECT, CLIENT_FACTORY[clientId] still points at the closed channel, so the next CONNECT with that id gets previous != null, logs "MQTT clientId {} is already connected, closing the previous connection." at INFO, and calls previous.close() + WillRepository.remove(previous) on a dead channel. Behaviour is harmless, but on a gateway where clients reconnect on a timer this fires constantly and will be read as a real takeover by whoever debugs it.
3. isOwner reports true for a dead channel. The Javadoc says "true when the channel is the current owner of the client id", but after a normal disconnect the closed channel remains the recorded owner forever.

Fix (note the two-argument remove, which is the point):

@Override
public void remove(final Channel channel) {
    final String clientId = CHANNEL_FACTORY.remove(channel);
    if (Objects.nonNull(clientId)) {
        // only clear ownership if this channel still holds it: a stale channel
        // closing late must not evict the connection that took over (#7918)
        CLIENT_FACTORY.remove(clientId, channel);
    }
}

A plain CLIENT_FACTORY.remove(clientId) here would be the same class of bug this PR is fixing — the stale channel would evict the new owner — so the conditional two-arg form matters.

add() is now dead in production but still writes the ownership index

Connect was the only production caller of add, and it now calls takeover instead (Subscribe / Unsubscribe only use get, Disconnect uses get / isOwner / remove). So add survives only to satisfy BaseRepository, while still doing CLIENT_FACTORY.put(clientId, channel) — a non-atomic second write that bypasses takeover's synchronized block. Either drop the CLIENT_FACTORY.put from add so the ownership index is written by takeover alone, or delete add from the interface if nothing needs it. Right now the invariant "ownership is only transferred by takeover" is not enforced by the code.

Non-blocking

• Discarding the replaced connection's will is a deliberate deviation from other brokers. The PR cites MQTT-3.1.4 for closing the existing connection, which is correct, but the spec does not say the will must be dropped; EMQX and Mosquitto both publish the will when a client is taken over, because from the old connection's point of view it was closed without a DISCONNECT. Dropping it is defensible for reconnect storms, but it changes observable behaviour and deserves a note in the class Javadoc (or a config switch) so it is not mistaken for a spec-mandated choice.
• The test cannot catch the leak. CHANNEL_FACTORY / CLIENT_FACTORY are static final, so new ChannelRepository() in @BeforeEach gives no isolation, and tearDown has no way to clear CLIENT_FACTORY. A test that connects, disconnects cleanly, and then asserts channelRepository.isOwner(CLIENT_ID, channel) is false (or that a subsequent CONNECT produces no takeover log) would have caught the issue above and pins the fix.
• takeover is synchronized on the repository instance, which works because Singleton holds a single instance — but CLIENT_FACTORY being static while the lock is per-instance is a mismatch. Making the maps instance fields would align them; the test's new ChannelRepository() would then also give real isolation.

The three tests are good — testStaleChannelCleanupKeepsTheNewSession reproduces the issue faithfully — and the EmbeddedChannel + real Connect setup is the right level of fidelity. Just need the ownership index cleaned up.

[konlue]

All addressed:

1. Ownership index cleanup: remove now clears the ownership index with the conditional two-arg remove, so a clean disconnect releases the id and a stale channel closing late cannot evict its replacement — pinned by a new test asserting isOwner turns false after a clean disconnect.
2. Ownership transfer centralized: add no longer writes the ownership index; ownership is transferred by takeover alone (DisconnectDispatchTest now registers through takeover, matching production).

Also took the non-blocking points:

• The maps are instance fields now, so the lock and the state share a lifetime and new ChannelRepository() gives real test isolation.
• The class Javadoc documents that dropping the replaced connection's will on takeover is a deliberate deviation from EMQX/Mosquitto behavior.
• The takeover-before-close ordering you verified now carries a comment at the call site.

[Aias00]

Review: #7939 fix(mqtt): take over duplicate clientIds and fence discarded clients (re-review)

Verdict: approved — cleanly closes the ownership leak and ensures robust, bounded MQTT client session takeover.

What was verified in this update

1. Safe ownership index cleanup:
   • remove(channel) now retrieves clientId and performs a conditional two-argument removal:

     clientFactory.remove(clientId, channel);

     This prevents memory leaks on disconnects while ensuring a stale connection closing late cannot evict a newer session that already took over the client ID.
2. Atomic ownership mutations:
   • Dropped clientFactory.put from add(...), ensuring that clientFactory is modified exclusively through the synchronized takeover(...) entry point.
3. Instance-level isolation:
   • Refactored channelFactory and clientFactory from static final to instance fields (final Map<...> initialized to ConcurrentHashMap), aligning map locks with takeover(...) synchronization and providing true test isolation.
4. Test coverage:
   • Added testCleanDisconnectReleasesOwnership, verifying that a clean disconnect clears client ownership and prevents spurious takeover warnings on subsequent connects.