fix: set MQTT sessionPresent from cleanSession flag and stored session state (#6889)

[lll-peanut]

fix: #6889

• Connect no longer hard-codes sessionPresent(true); it is derived from the cleanSession flag and stored per-client session state
• add SessionRepository/MqttSession and resume stored subscriptions on reconnect
• add unit tests covering first connect, reconnect, cleanSession discard, and subscription resume

DISCONNECT is now dispatched by MqttFactory, so clean-session sessions are removed on graceful disconnect; the disconnected channel is also removed from SubscribeRepository while persistent session state is kept for resume.

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.

[Aias00]

Good fix — the hardcoded sessionPresent(true) was a clear spec violation (MQTT-3.2.2-6/7), and the new derivation !cleanSession && session != null is correct, including the ordering (sessionPresent computed before the cleanSession overwrite). Two things worth addressing:

Subscription resume loses QoS. Connect.java (~L91) calls SubscribeRepository.add(new ArrayList<>(session.getTopics()), Collections.singletonList(ctx.channel())), which goes through the add(List<String>, List<Channel>) overload (SubscribeRepository.java:43) — a topic→channel mapping only. It bypasses the QoS-aware add(Channel, List<MqttTopicSubscription>) overload (L56). MqttSession.topics is a Set<String>, so the original QoS levels (and duplicate subscriptions with differing QoS) are collapsed; resumed subscriptions get an implicit default QoS rather than the client's original. Could MqttSession store List<MqttTopicSubscription> (or Map<String, MqttQoS>) and resume via the QoS-aware overload?

Clean-session state leaks on graceful DISCONNECT (dead-code cleanup). As the PR body notes, Disconnect isn't dispatched by MqttFactory, so the cleanup added in Disconnect.java:42-52 is unreachable today. Consequence: a clean-session client that disconnects gracefully leaves its MqttSession in SessionRepository until the same clientId next connects clean (Connect.java discards it lazily); clients that never reconnect leak indefinitely. Acceptable as a follow-up, but please either wire Disconnect dispatch in this PR or open a tracking issue and reference it in the PR/code so the leak isn't lost.

Minor (pre-existing, now more reachable): SubscribeRepository.add(List,List) (L43-49) has a check-then-act race — get(s) returns getOrDefault(topic, new CopyOnWriteArrayList<>()), so for an absent topic two concurrent resumes can each create a fresh list and one put overwrites the other's channel. Not introduced here, just noting the resume path makes it reachable.

[lll-peanut]

Good fix — the hardcoded sessionPresent(true) was a clear spec violation (MQTT-3.2.2-6/7), and the new derivation !cleanSession && session != null is correct, including the ordering (sessionPresent computed before the cleanSession overwrite). Two things worth addressing:

Subscription resume loses QoS. Connect.java (~L91) calls SubscribeRepository.add(new ArrayList<>(session.getTopics()), Collections.singletonList(ctx.channel())), which goes through the add(List<String>, List<Channel>) overload (SubscribeRepository.java:43) — a topic→channel mapping only. It bypasses the QoS-aware add(Channel, List<MqttTopicSubscription>) overload (L56). MqttSession.topics is a Set<String>, so the original QoS levels (and duplicate subscriptions with differing QoS) are collapsed; resumed subscriptions get an implicit default QoS rather than the client's original. Could MqttSession store List<MqttTopicSubscription> (or Map<String, MqttQoS>) and resume via the QoS-aware overload?

Clean-session state leaks on graceful DISCONNECT (dead-code cleanup). As the PR body notes, Disconnect isn't dispatched by MqttFactory, so the cleanup added in Disconnect.java:42-52 is unreachable today. Consequence: a clean-session client that disconnects gracefully leaves its MqttSession in SessionRepository until the same clientId next connects clean (Connect.java discards it lazily); clients that never reconnect leak indefinitely. Acceptable as a follow-up, but please either wire Disconnect dispatch in this PR or open a tracking issue and reference it in the PR/code so the leak isn't lost.

Minor (pre-existing, now more reachable): SubscribeRepository.add(List,List) (L43-49) has a check-then-act race — get(s) returns getOrDefault(topic, new CopyOnWriteArrayList<>()), so for an absent topic two concurrent resumes can each create a fresh list and one put overwrites the other's channel. Not introduced here, just noting the resume path makes it reachable.

Thanks for the review. I addressed the three points:

1. MqttSession now stores subscriptions as topic -> MqttQoS, and Connect resumes stored subscriptions through the QoS-aware SubscribeRepository.add(Channel,
   List) path. I also added tests for QoS retention and QoS replacement on resubscribe.

2. DISCONNECT is now dispatched by MqttFactory. Clean-session DISCONNECT removes the stored session, and DISCONNECT also unregisters the current channel from SubscribeRepository
   while keeping persistent session state for later resume.

3. SubscribeRepository.add now uses computeIfAbsent instead of the previous get/default-list/put check-then-act sequence, including the QoS-aware overload.

I also added a guard return after rejecting unsupported protocol versions to avoid continuing through authentication/session setup after sending the CONNACK rejection.