QueueImpl.depage() acquires {@code depageLock} then {@code synchronized(this)}. + * QueueImpl.deleteReference() is {@code synchronized(this)}, then calls {@code iterQueue()} + * which acquires {@code depageLock}. Concurrent execution causes a classic ABBA deadlock. + * + *
Bug introduced in commit e9dbc11 (ARTEMIS-5376, Feb 27 2026). First affected release: 2.53.0.
+ */
+public class DeadlockDepageLockDeleteReferenceTest extends ActiveMQTestBase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final String ADDRESS = "test.deadlock.address";
+ private static final String QUEUE = "test.deadlock.queue";
+
+ // Small address limit so paging kicks in after ~50 messages.
+ private static final int PAGE_SIZE_BYTES = 128 * 1024; // 128 KB per page file
+ private static final int MAX_SIZE_BYTES = 512 * 1024; // 512 KB address limit
+ private static final int MESSAGE_SIZE = 10 * 1024; // 10 KB per message
+ // 100 × 10 KB = 1 MB >> 512 KB; keeps ~50 messages paged to disk.
+ private static final int INITIAL_MESSAGES = 100;
+
+ private static final long DEADLOCK_TIMEOUT_MS = 10_000;
+
+ private ServerLocator locator;
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ locator = createInVMNonHALocator();
+ }
+
+ @Test
+ @Timeout(120)
+ public void testNoDeadlockDuringDepageAndDeleteReference() throws Exception {
+ ActiveMQServer server = createServer(true, createDefaultInVMConfig(), PAGE_SIZE_BYTES, MAX_SIZE_BYTES);
+ server.start();
+
+ server.createQueue(QueueConfiguration.of(QUEUE)
+ .setAddress(ADDRESS)
+ .setRoutingType(RoutingType.ANYCAST)
+ .setDurable(true));
+
+ QueueControl queueControl = (QueueControl) server.getManagementService()
+ .getResource(ResourceNames.QUEUE + QUEUE);
+
+ ClientSessionFactory factory = locator.createSessionFactory();
+
+ // Fill queue past address limit to trigger paging.
+ try (ClientSession sendSession = factory.createSession(false, true, true)) {
+ ClientProducer producer = sendSession.createProducer(ADDRESS);
+ byte[] body = new byte[MESSAGE_SIZE];
+ Arrays.fill(body, (byte) 'X');
+ for (int i = 0; i < INITIAL_MESSAGES; i++) {
+ ClientMessage msg = sendSession.createMessage(true);
+ msg.getBodyBuffer().writeBytes(body);
+ producer.send(msg);
+ }
+ }
+
+ // Background sender: keeps paging active by replenishing faster than the consumer drains.
+ ClientSession senderSession = factory.createSession(false, true, true);
+ ClientProducer bgProducer = senderSession.createProducer(ADDRESS);
+ byte[] bgBody = new byte[MESSAGE_SIZE];
+ Arrays.fill(bgBody, (byte) 'Y');
+ Thread senderThread = new Thread(() -> {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ ClientMessage msg = senderSession.createMessage(true);
+ msg.getBodyBuffer().writeBytes(bgBody);
+ bgProducer.send(msg);
+ Thread.sleep(10); // ~100 msg/s >> consumer rate, keeps paging active
+ }
+ } catch (InterruptedException ignored) {
+ } catch (Exception e) {
+ if (!Thread.currentThread().isInterrupted()) {
+ logger.warn("sender error: {}", e.getMessage());
+ }
+ }
+ }, "repro-sender");
+ senderThread.setDaemon(true);
+ senderThread.start();
+
+ // Consumer: triggers continuous depage() (acquires depageLock, then synchronized(this)).
+ ClientSession consumeSession = factory.createSession(false, false, false);
+ consumeSession.start();
+ ClientConsumer consumer = consumeSession.createConsumer(QUEUE);
+ Thread consumerThread = new Thread(() -> {
+ try {
+ int batch = 0;
+ while (!Thread.currentThread().isInterrupted()) {
+ ClientMessage msg = consumer.receive(200);
+ if (msg != null) {
+ msg.acknowledge();
+ if (++batch % 10 == 0) {
+ consumeSession.commit();
+ }
+ Thread.sleep(100); // ~10 msg/s; sender at 100 msg/s keeps queue growing
+ }
+ }
+ } catch (InterruptedException ignored) {
+ } catch (Exception e) {
+ if (!Thread.currentThread().isInterrupted()) {
+ logger.warn("consumer error: {}", e.getMessage());
+ }
+ }
+ }, "repro-consumer");
+ consumerThread.setDaemon(true);
+ consumerThread.start();
+
+ AtomicBoolean stop = new AtomicBoolean(false);
+ ExecutorService pool = Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ return t;
+ });
+
+ // Management threads: removeMessage() -> deleteReference() -> iterQueue() -> depageLock.
+ // deleteReference() holds synchronized(QueueImpl.this), then acquires depageLock inside
+ // iterQueue(). This is the opposite order from depage(), causing the ABBA deadlock.
+ for (int i = 0; i < 5; i++) {
+ pool.submit(() -> {
+ while (!stop.get()) {
+ try {
+ @SuppressWarnings("unchecked")
+ Map