You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@artemis.apache.org by jb...@apache.org on 2026/06/01 16:30:46 UTC
(artemis) branch main updated: ARTEMIS-6085 Fix NPE from race in MQTTStateManager.getSessionState
This is an automated email from the ASF dual-hosted git repository.
jbertram pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/artemis.git
The following commit(s) were added to refs/heads/main by this push:
new cd396d02cb ARTEMIS-6085 Fix NPE from race in MQTTStateManager.getSessionState
cd396d02cb is described below
commit cd396d02cbe66aeea2a585a4f272183676e8dc97
Author: Karol <ka...@gmail.com>
AuthorDate: Sat May 30 09:07:04 2026 +0200
ARTEMIS-6085 Fix NPE from race in MQTTStateManager.getSessionState
getSessionState() used a non-atomic containsKey()/get() pair on a
ConcurrentHashMap. During link stealing (two connections sharing a
client ID) removeSessionState() can run between the two calls, so get()
returns null and handleLinkStealing() dereferences it via .getSession(),
throwing a NullPointerException.
Use computeIfAbsent() for an atomic get-or-create that never returns null.
Adds a concurrency regression test that reproduces the NPE.
---
artemis-protocols/artemis-mqtt-protocol/pom.xml | 5 +
.../core/protocol/mqtt/MQTTStateManager.java | 12 +--
.../core/protocol/mqtt/MQTTStateManagerTest.java | 107 +++++++++++++++++++++
3 files changed, 116 insertions(+), 8 deletions(-)
diff --git a/artemis-protocols/artemis-mqtt-protocol/pom.xml b/artemis-protocols/artemis-mqtt-protocol/pom.xml
index 62c575526c..ccb22c3195 100644
--- a/artemis-protocols/artemis-mqtt-protocol/pom.xml
+++ b/artemis-protocols/artemis-mqtt-protocol/pom.xml
@@ -107,6 +107,11 @@
<artifactId>junit-jupiter-engine</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j2-impl</artifactId>
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 69c2101572..99e3d2265f 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
@@ -137,14 +137,10 @@ public class MQTTStateManager {
public MQTTSessionState getSessionState(String clientId) throws Exception {
// [MQTT-3.1.2-4] Attach an existing session if one exists otherwise create a new one.
- if (sessionStates.containsKey(clientId)) {
- return sessionStates.get(clientId);
- } else {
- MQTTSessionState sessionState = new MQTTSessionState(clientId);
- logger.debug("Adding MQTT session state for: {}", clientId);
- sessionStates.put(clientId, sessionState);
- return sessionState;
- }
+ return sessionStates.computeIfAbsent(clientId, key -> {
+ logger.debug("Adding MQTT session state for: {}", key);
+ return new MQTTSessionState(key);
+ });
}
public MQTTSessionState removeSessionState(String clientId) throws Exception {
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManagerTest.java b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManagerTest.java
new file mode 100644
index 0000000000..0b4b461ce7
--- /dev/null
+++ b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManagerTest.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.activemq.artemis.core.protocol.mqtt;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class MQTTStateManagerTest {
+
+ @Test
+ @Timeout(60)
+ public void testGetSessionStateNeverReturnsNullUnderConcurrentRemoval() throws Exception {
+ final ActiveMQServer server = mock(ActiveMQServer.class);
+ final Configuration configuration = mock(Configuration.class);
+ when(server.getConfiguration()).thenReturn(configuration);
+ when(configuration.isMqttSubscriptionPersistenceEnabled()).thenReturn(false);
+
+ final MQTTStateManager manager = MQTTStateManager.getInstance(server);
+ try {
+ final String clientId = "link-stealing-client";
+ final int iterations = 2_000_000;
+ final int removerThreads = 3;
+
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+ final AtomicBoolean stop = new AtomicBoolean(false);
+ final CountDownLatch start = new CountDownLatch(1);
+ final ExecutorService pool = Executors.newFixedThreadPool(removerThreads + 1);
+
+ for (int i = 0; i < removerThreads; i++) {
+ pool.submit(() -> {
+ awaitQuietly(start);
+ while (!stop.get() && failure.get() == null) {
+ try {
+ manager.removeSessionState(clientId);
+ } catch (Throwable t) {
+ failure.compareAndSet(null, t);
+ return;
+ }
+ }
+ });
+ }
+
+ pool.submit(() -> {
+ awaitQuietly(start);
+ try {
+ for (int i = 0; i < iterations && failure.get() == null; i++) {
+ MQTTSessionState state = manager.getSessionState(clientId);
+ assertNotNull(state, "getSessionState(String) must never return null (ARTEMIS-6085)");
+ }
+ } catch (Throwable t) {
+ failure.compareAndSet(null, t);
+ } finally {
+ stop.set(true);
+ }
+ });
+
+ start.countDown();
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(50, TimeUnit.SECONDS), "concurrency test did not finish in time");
+
+ if (failure.get() != null) {
+ fail("getSessionState() returned null / threw under concurrent removal (ARTEMIS-6085)", failure.get());
+ }
+ } finally {
+ MQTTStateManager.removeInstance(server);
+ }
+ }
+
+ private static void awaitQuietly(CountDownLatch latch) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@artemis.apache.org
For additional commands, e-mail: commits-help@artemis.apache.org