From 95a0be330bb65269a86efc709860dc515122b2d4 Mon Sep 17 00:00:00 2001 From: Shivan Taher Date: Tue, 11 Aug 2026 10:20:30 +0200 Subject: [PATCH] ARTEMIS-6179 Fix deleteReference/depage deadlock Removes synchronized from QueueImpl#deleteReference(). It calls iterQueue(), which acquires depageLock internally, while depage() acquires depageLock first and then enters a synchronized(this) block. The reversed lock order deadlocks QueueControl#removeMessage() against the paging executor whenever they race on an actively paging queue. This mirrors the fix already applied to copyReference() in ARTEMIS-5376: iterQueue() already provides its own synchronization via depageLock, so the outer synchronized on deleteReference() is redundant and unsafe. Verified with testRemoveMessageWhilstPagingAndConsuming, which reliably deadlocks without this change and passes cleanly with it. --- .../artemis/core/server/impl/QueueImpl.java | 4 +- .../ManagementWithPagingServerTest.java | 325 ++++++++++++------ 2 files changed, 230 insertions(+), 99 deletions(-) diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java index 04a47ad3aa4..4d70d3f6cf1 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java @@ -2077,7 +2077,7 @@ private int iterQueue(final String operationName, try { Transaction tx = new TransactionImpl(storageManager); - synchronized (this) { + synchronized (QueueImpl.this) { // ensure all messages are moved from intermediateMessageReferences so that they can be seen by the iterator doInternalPoll(); @@ -2223,7 +2223,7 @@ public void destroyPaging() throws Exception { } @Override - public synchronized boolean deleteReference(final long messageID) throws Exception { + public boolean deleteReference(final long messageID) throws Exception { return iterQueue("deleteReference", DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) { @Override public boolean actMessage(Transaction tx, MessageReference ref) throws Exception { diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java index 09ddd9006bc..d421fd8d8a7 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java @@ -20,17 +20,28 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.core.server.Queue; import org.apache.activemq.artemis.json.JsonArray; import org.apache.activemq.artemis.json.JsonNumber; import org.apache.activemq.artemis.json.JsonObject; import org.apache.activemq.artemis.json.JsonValue; + +import java.lang.invoke.MethodHandles; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.JsonUtil; @@ -51,16 +62,21 @@ import org.apache.activemq.artemis.utils.RandomUtil; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * This class contains tests for core management functionalities that are affected by a server in paging mode. */ public class ManagementWithPagingServerTest extends ManagementTestBase { + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + private ActiveMQServer server; private ClientSession session1; private ClientSession session2; private ServerLocator locator; + private ClientSessionFactory sf; @Test public void testListMessagesAsJSON() throws Exception { @@ -72,15 +88,16 @@ public void testListMessagesAsJSON() throws Exception { QueueControl queueControl = createManagementControl(address, queue); int num = 1000; - SenderThread sender = new SenderThread(address, num, 0); - - ReceiverThread receiver = new ReceiverThread(queue, num, 0); + SenderRunnable sender = new SenderRunnable(address, num, 0); + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 0); - //kick off sender - sender.start(); + ExecutorService executorService = Executors.newFixedThreadPool(1); + runAfter(executorService::shutdownNow); + runAfter(sender::stop); + runAfter(receiver::stop); - //wait for all messages sent - sender.join(); + executorService.execute(sender); + sender.waitDone(); assertNull(sender.getError()); long count = queueControl.countMessages(null); @@ -98,9 +115,8 @@ public void testListMessagesAsJSON() throws Exception { } assertEquals(num, array.size()); - //kick off receiver - receiver.start(); - receiver.join(); + executorService.execute(receiver); + receiver.waitDone(); assertNull(receiver.getError()); result = queueControl.listMessagesAsJSON(null); @@ -154,10 +170,12 @@ public void testListMessagesAsJSONWithFilter() throws Exception { long n = queueControl.countMessages(filter); assertEquals(num / 2, n); - //drain out messages - ReceiverThread receiver = new ReceiverThread(queue, num, 1); - receiver.start(); - receiver.join(); + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 1); + ExecutorService executorService = Executors.newFixedThreadPool(1); + runAfter(executorService::shutdownNow); + runAfter(receiver::stop); + executorService.execute(receiver); + receiver.waitDone(); } //In this test, the management api listMessageAsJSon is called while @@ -174,30 +192,29 @@ public void testListMessagesAsJSONWhilePagingOnGoing() throws Exception { QueueControl queueControl = createManagementControl(address, queue); int num = 1000; - SenderThread sender = new SenderThread(address, num, 1); + SenderRunnable sender = new SenderRunnable(address, num, 1); + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 2); + ManagementRunnable console = new ManagementRunnable(queueControl); - ReceiverThread receiver = new ReceiverThread(queue, num, 2); + ExecutorService executorService = Executors.newFixedThreadPool(2); + runAfter(executorService::shutdownNow); + runAfter(sender::stop); + runAfter(receiver::stop); + runAfter(console::stop); - ManagementThread console = new ManagementThread(queueControl); + executorService.execute(sender); + executorService.execute(console); - //kick off sender - sender.start(); - - //kick off jmx client - console.start(); - - //wait for all messages sent - sender.join(); + sender.waitDone(); assertNull(sender.getError()); - //kick off receiver - receiver.start(); + executorService.execute(receiver); - receiver.join(); + receiver.waitDone(); assertNull(receiver.getError()); - console.exit(); - console.join(); + console.stop(); + console.waitDone(); assertNull(console.getError()); } @@ -300,7 +317,7 @@ public void testMoveMessageWhilstPagingAndConsuming() throws Exception { QueueControl otherQueueControl = createManagementControl(otherAddress, otherQueue); - int num = 1000; + int num = 10; ClientProducer producer = session1.createProducer(address); for (int i = 0; i < num; i++) { @@ -308,20 +325,92 @@ public void testMoveMessageWhilstPagingAndConsuming() throws Exception { producer.send(message); } - ManagementCopyThread console = new ManagementCopyThread(queueControl, otherQueue.toString()); - ReceiverThread receiver = new ReceiverThread(queue, num, 0); - console.start(); - receiver.start(); + ExecutorService executorService = Executors.newFixedThreadPool(2); + runAfter(executorService::shutdownNow); - receiver.join(); - console.stop = true; - console.join(); + ManagementCopyRunnable console = new ManagementCopyRunnable(queue.toString(), queueControl, otherQueue.toString()); + runAfter(console::stop); + + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 0); + runAfter(receiver::stop); + executorService.execute(receiver); + executorService.execute(console); + + receiver.waitDone(); + + console.stop(); + console.waitDone(); Map[] messages = otherQueueControl.listMessages(null); assertEquals(messages.length, console.copiedMessages); } + /** + * Reproduction for ARTEMIS-6179: QueueImpl#deleteReference() is still {@code synchronized} and calls + * iterQueue(), which locks depageLock -- while QueueImpl#depage() locks depageLock first and then enters a + * synchronized(this) block. Racing QueueControl#removeMessage() (-> deleteReference()) against depaging + * (triggered here by the ReceiverThread acking messages) can deadlock the two threads against each other. + * This mirrors testMoveMessageWhilstPagingAndConsuming(), which caught the equivalent bug for copyReference() + * (ARTEMIS-5376), replacing copyMessage with removeMessage. Detection uses the JVM's own deadlock detector + * (ThreadMXBean) instead of a fixed timeout, since a hang here is the failure itself. + */ + @Test + public void testRemoveMessageWhilstPagingAndConsuming() throws Exception { + final int messagesPerIteration = 100; + final int maxIterations = 2; + + SimpleString address = RandomUtil.randomUUIDSimpleString(); + SimpleString queue = RandomUtil.randomUUIDSimpleString(); + + session1.createQueue(QueueConfiguration.of(queue).setAddress(address)); + + QueueControl queueControl = createManagementControl(address, queue); + + ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean(); + + // A single remover is sufficient: QueueControlImpl#removeMessage() is serialized broker-wide through + // ActiveMQServerImpl's managementLock, so extra concurrent removers would just contend on that lock + // without increasing the odds of racing depage(). + ManagementRemoveThread remover = new ManagementRemoveThread(queueControl); + + ExecutorService executorService = Executors.newFixedThreadPool(3); + runAfter(executorService::shutdownNow); + runAfter(remover::stop); + + executorService.execute(remover); + + long[] deadlockedIds = null; + + for (int iteration = 0; iteration < maxIterations; iteration++) { + logger.info("iteration : {}", iteration); + SenderRunnable sender = new SenderRunnable(address, messagesPerIteration, 0); + ReceiverRunnable receiver = new ReceiverRunnable(queue, messagesPerIteration, 0); + + executorService.execute(sender); + executorService.execute(receiver); + + sender.waitDone(); + receiver.waitDone(); + + deadlockedIds = threadMXBean.findDeadlockedThreads(); + + if (deadlockedIds != null) { + StringBuilder sb = new StringBuilder("Deadlock detected between removeMessage() and depage():\n"); + for (long id : deadlockedIds) { + ThreadInfo info = threadMXBean.getThreadInfo(id, Integer.MAX_VALUE); + sb.append(info).append('\n'); + } + fail(sb.toString()); + } + } + + remover.stop(); + remover.waitDone(); + assertNull(remover.getError()); + + } + @Override @BeforeEach public void setUp() throws Exception { @@ -338,84 +427,115 @@ public void setUp() throws Exception { server.start(); locator = createInVMNonHALocator().setBlockOnNonDurableSend(false).setConsumerWindowSize(0); - ClientSessionFactory sf = createSessionFactory(locator); + sf = createSessionFactory(locator); session1 = sf.createSession(false, true, false); session1.start(); session2 = sf.createSession(false, true, false); session2.start(); } - private class SenderThread extends Thread { + protected abstract class AbstractRunnable implements Runnable { + protected volatile boolean stop = false; + protected volatile Exception error = null; + protected final CountDownLatch done = new CountDownLatch(1); + + public boolean waitDone() throws Exception { + return done.await(10, TimeUnit.SECONDS); + } + + protected void done() { + done.countDown(); + } + + public void stop() { + stop = true; + } + + public Exception getError() { + return this.error; + } + } + + private class SenderRunnable extends AbstractRunnable { private SimpleString address; private int num; private long delay; private volatile Exception error = null; + ClientSession sessionSender; - private SenderThread(SimpleString address, int num, long delay) { + private SenderRunnable(SimpleString address, int num, long delay) throws Exception { this.address = address; this.num = num; this.delay = delay; + sessionSender = sf.createSession(false, false, false); } @Override public void run() { - ClientProducer producer; + try { + ClientProducer producer; - byte[] body = new byte[128]; - ByteBuffer bb = ByteBuffer.wrap(body); - for (int j = 1; j <= 128; j++) { - bb.put(getSamplebyte(j)); - } + byte[] body = new byte[128]; + ByteBuffer bb = ByteBuffer.wrap(body); + for (int j = 1; j <= 128; j++) { + bb.put(getSamplebyte(j)); + } - try { - producer = session1.createProducer(address); - - for (int i = 0; i < num; i++) { - ClientMessage message = session1.createMessage(true); - message.setPriority((byte) 1); - ActiveMQBuffer buffer = message.getBodyBuffer(); - buffer.writeBytes(body); - producer.send(message); - try { - Thread.sleep(delay); - } catch (InterruptedException e) { - //ignore + try { + producer = sessionSender.createProducer(address); + + for (int i = 0; i < num && !stop; i++) { + ClientMessage message = sessionSender.createMessage(true); + message.setPriority((byte) 1); + ActiveMQBuffer buffer = message.getBodyBuffer(); + buffer.writeBytes(body); + producer.send(message); + if ((i + 1) % 100 == 0) { + sessionSender.commit(); + } + try { + Thread.sleep(delay); + } catch (InterruptedException e) { + //ignore + } } + sessionSender.commit(); + } catch (Exception e) { + error = e; } - } catch (Exception e) { - error = e; + } finally { + done(); } } - public Exception getError() { - return this.error; - } } - private class ReceiverThread extends Thread { + private class ReceiverRunnable extends AbstractRunnable { private SimpleString queue; private int num; private long delay; - private volatile Exception error = null; + private ClientSession sessionConsumer; - private ReceiverThread(SimpleString queue, int num, long delay) { + private ReceiverRunnable(SimpleString queue, int num, long delay) throws Exception { this.queue = queue; this.num = num; this.delay = delay; + this.sessionConsumer = sf.createSession(false, true, false); } @Override public void run() { ClientConsumer consumer; try { - consumer = session2.createConsumer(queue); + consumer = sessionConsumer.createConsumer(queue); + sessionConsumer.start(); - for (int i = 0; i < num; i++) { + for (int i = 0; i < num && !stop; i++) { ClientMessage message = consumer.receive(5000); message.acknowledge(); - session2.commit(); + sessionConsumer.commit(); try { Thread.sleep(delay); } catch (InterruptedException e) { @@ -424,21 +544,17 @@ public void run() { } } catch (Exception e) { error = e; + } finally { + done(); } } - - public Exception getError() { - return this.error; - } } - private class ManagementThread extends Thread { + private class ManagementRunnable extends AbstractRunnable { private QueueControl queueControl; - private volatile boolean stop = false; - private Exception error = null; - private ManagementThread(QueueControl queueControl) { + private ManagementRunnable(QueueControl queueControl) { this.queueControl = queueControl; } @@ -456,55 +572,70 @@ public void run() { } } catch (Exception e) { error = e; + } finally { + done(); } } - - public Exception getError() { - return error; - } - - public void exit() { - stop = true; - } } - private class ManagementCopyThread extends Thread { + private class ManagementCopyRunnable extends AbstractRunnable { private QueueControl queueControl; private String queue; - private volatile boolean stop = false; + private String originalQueue; + Queue targetQueue; int copiedMessages = 0; - private Exception error = null; - private ManagementCopyThread(QueueControl queueControl, String queue) { + private ManagementCopyRunnable(String originalQueue, QueueControl queueControl, String queue) { this.queueControl = queueControl; this.queue = queue; + this.originalQueue = originalQueue; + targetQueue = server.locateQueue(queue); } @Override public void run() { try { Random random = new Random(System.currentTimeMillis()); - while (!stop) { + while (!stop && copiedMessages < 100) { long messageID = random.nextInt(1000); boolean copied = queueControl.copyMessage(messageID, queue); - System.out.println("messageID = " + messageID); if (copied) { copiedMessages++; + logger.info("Copied message ID {}, totalCopied so far = {}", messageID, copiedMessages); + targetQueue.forEach(r -> { + logger.info("containing {}", r.getMessage()); + }); } } } catch (Exception e) { error = e; + } finally { + done(); } } + } - public Exception getError() { - return error; + private class ManagementRemoveThread extends AbstractRunnable { + + private QueueControl queueControl; + + private ManagementRemoveThread(QueueControl queueControl) { + this.queueControl = queueControl; } - public void exit() { - stop = true; + @Override + public void run() { + try { + while (!stop) { + queueControl.removeMessage(0); + } + } catch (Exception e) { + error = e; + } finally { + done(); + } } } }