From 31ef155e5779fed793b9a564901812f18f3ce175 Mon Sep 17 00:00:00 2001
From: Alexey Markevich <buhhunyx@gmail.com>
Date: Thu, 1 Oct 2026 11:08:31 +0200
Subject: [PATCH] ARTEMIS-6286 Use concurrent map for MQTTStateManager
 instances

---
 .../core/protocol/mqtt/MQTTStateManager.java  | 26 ++++++-------------
 1 file changed, 8 insertions(+), 18 deletions(-)

diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java
index 0ff12ba598b..539a6686fb5 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java
@@ -20,7 +20,7 @@
 
 import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
@@ -50,7 +50,7 @@
 public class MQTTStateManager {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-   private static final Map<Integer, MQTTStateManager> INSTANCES = new HashMap<>();
+   private static final Map<Integer, MQTTStateManager> INSTANCES = new ConcurrentHashMap<>();
    private final ActiveMQServer server;
    private final Map<String, MQTTSessionState> sessionStates = new ConcurrentHashMap<>();
    private Queue sessionStore;
@@ -112,21 +112,15 @@ void clear(String clientId) {
     * one instance of MQTTSessionStateManager per-broker with the understanding that there can be multiple brokers in
     * the same JVM.
     */
-   public static synchronized MQTTStateManager getInstance(ActiveMQServer server) throws Exception {
-      MQTTStateManager instance = INSTANCES.get(System.identityHashCode(server));
-      if (instance == null) {
-         instance = new MQTTStateManager(server);
-         INSTANCES.put(System.identityHashCode(server), instance);
-      }
-
-      return instance;
+   public static MQTTStateManager getInstance(ActiveMQServer server) {
+      return INSTANCES.computeIfAbsent(System.identityHashCode(server), id -> new MQTTStateManager(server));
    }
 
-   public static synchronized void removeInstance(ActiveMQServer server) {
+   public static void removeInstance(ActiveMQServer server) {
       INSTANCES.remove(System.identityHashCode(server));
    }
 
-   private MQTTStateManager(ActiveMQServer server) throws Exception {
+   private MQTTStateManager(ActiveMQServer server) {
       this.server = server;
       this.subscriptionPersistenceEnabled = server.getConfiguration().isMqttSubscriptionPersistenceEnabled();
       this.journalHashMapProvider = new JournalHashMapProvider<>(server.getStorageManager()::generateID, server.getStorageManager(), PacketIdCorrelationKey.getPersister(), JournalRecordIds.MQTT_PACKET_ID_CORRELATION, OperationContextImpl::getContext, null, server.getIoCriticalErrorListener());
@@ -172,11 +166,7 @@ public MQTTSessionState getSessionState(String clientId) throws Exception {
 
    public MQTTSessionState removeSessionState(String clientId) throws Exception {
       logger.debug("Removing MQTT session state for: {}", clientId);
-      if (clientId == null) {
-         return null;
-      }
-      MQTTSessionState removed = sessionStates.remove(clientId);
-      return removed;
+      return clientId != null ? sessionStates.remove(clientId) : null;
    }
 
    public void removeDurableSubscriptionState(String clientId) throws Exception {
@@ -187,7 +177,7 @@ public void removeDurableSubscriptionState(String clientId) throws Exception {
    }
 
    public Map<String, MQTTSessionState> getSessionStates() {
-      return new HashMap<>(sessionStates);
+      return Collections.unmodifiableMap(sessionStates);
    }
 
    @Override