You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@artemis.apache.org by cl...@apache.org on 2026/08/13 17:19:07 UTC
(artemis) branch main updated: ARTEMIS-6179 Fix deleteReference/depage deadlock
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic 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 00dd2dde19 ARTEMIS-6179 Fix deleteReference/depage deadlock
00dd2dde19 is described below
commit 00dd2dde19f974bf1ee555881f132f995045b95a
Author: Shivan Taher <sh...@oviva.com>
AuthorDate: Tue Aug 11 10:20:30 2026 +0200
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 +-
.../management/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 04a47ad3aa..4d70d3f6cf 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 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
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 class QueueImpl extends CriticalComponentImpl implements Queue {
}
@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 09ddd9006b..d421fd8d8a 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.assertEquals;
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.core.settings.impl.AddressSettings;
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 class ManagementWithPagingServerTest extends ManagementTestBase {
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 class ManagementWithPagingServerTest extends ManagementTestBase {
}
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 class ManagementWithPagingServerTest extends ManagementTestBase {
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 class ManagementWithPagingServerTest extends ManagementTestBase {
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 class ManagementWithPagingServerTest extends ManagementTestBase {
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 class ManagementWithPagingServerTest extends ManagementTestBase {
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<String, Object>[] 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 class ManagementWithPagingServerTest extends ManagementTestBase {
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 class ManagementWithPagingServerTest extends ManagementTestBase {
}
} 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 class ManagementWithPagingServerTest extends ManagementTestBase {
}
} 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();
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@artemis.apache.org
For additional commands, e-mail: commits-help@artemis.apache.org