Skip to content

ARTEMIS-6286 Use concurrent map for MQTTStateManager instances #6757

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jbertram merged 1 commit into from
Oct 6, 2026
Merged

ARTEMIS-6286 Use concurrent map for MQTTStateManager instances #6757

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 8 additions & 18 deletions ...otocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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 {
Expand All @@ -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
Expand Down
Loading