From 67ea634bb274a41fa00a3554dfa073e9ff90a165 Mon Sep 17 00:00:00 2001 From: grepjava Date: Sun, 8 Nov 2026 19:51:02 +0800 Subject: [PATCH] ARTEMIS-6362 A send refused after its duplicate id check leaves the duplicate id cache as it was --- .../impl/InMemoryDuplicateIDCache.java | 27 +++++++- .../impl/PersistentDuplicateIDCache.java | 27 +++++++- .../integration/DuplicateDetectionTest.java | 62 +++++++++++++++++++ 3 files changed, 110 insertions(+), 6 deletions(-) diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java index 8f012280c096..86203c9da10b 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java @@ -51,6 +51,10 @@ final class InMemoryDuplicateIDCache implements DuplicateIDCache { private final Map cache = new ConcurrentHashMap<>(); + // Ids of sends whose transaction is still open, with how many: they count as duplicates, but take no slot until + // the send commits, so a send that fails leaves the ids before it in the cache + private final Map inFlight = new ConcurrentHashMap<>(); + private final SimpleString address; private final List ids; @@ -123,7 +127,7 @@ public boolean contains(final byte[] duplID) { } private boolean contains(final ByteArray id) { - boolean contains = cache.containsKey(id); + boolean contains = cache.containsKey(id) || inFlight.containsKey(id); if (logger.isTraceEnabled()) { if (contains) { @@ -170,7 +174,7 @@ private synchronized void addToCache(final ByteArray holder, final Transaction t } if (instantAdd) { - addToCacheInMemory(holder); + reserve(holder); tx.addOperation(new AddDuplicateIDOperation(holder, false)); } else { // For a tx, it's important that the entry is not added to the cache until commit @@ -185,6 +189,14 @@ public void load(final Transaction tx, final byte[] duplID) { tx.addOperation(new AddDuplicateIDOperation(new ByteArray(duplID), true)); } + private void reserve(final ByteArray holder) { + inFlight.merge(holder, 1, Integer::sum); + } + + private void release(final ByteArray holder) { + inFlight.computeIfPresent(holder, (id, count) -> count == 1 ? null : count - 1); + } + private synchronized void addToCacheInMemory(final ByteArray holder) { if (logger.isTraceEnabled()) { logger.trace("address = {} adding {}", address, describeID(holder.bytes)); @@ -266,6 +278,15 @@ private void process() { } } + @Override + public void beforeCommit(final Transaction tx) throws Exception { + // The send was routed: its id takes its slot now, as a send outside a transaction does + if (!afterCommit) { + process(); + release(id); + } + } + @Override public void afterCommit(final Transaction tx) { if (afterCommit) { @@ -276,7 +297,7 @@ public void afterCommit(final Transaction tx) { @Override public void beforeRollback(Transaction tx) throws Exception { if (!afterCommit) { - deleteFromCache(id); + release(id); } } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java index 4d590fc62ff9..444e8b95ab7b 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java @@ -55,6 +55,10 @@ final class PersistentDuplicateIDCache implements DuplicateIDCache { private final Map cache = new ConcurrentHashMap<>(); + // Ids of sends whose transaction is still open, with how many: they count as duplicates, but take no slot until + // the send commits, so a send that fails leaves the ids before it in the cache + private final Map inFlight = new ConcurrentHashMap<>(); + private final SimpleString address; private final List> ids; @@ -201,7 +205,7 @@ public boolean contains(final byte[] duplID) { } private boolean contains(final ByteArray duplID) { - final boolean contains = cache.containsKey(duplID); + final boolean contains = cache.containsKey(duplID) || inFlight.containsKey(duplID); if (contains) { if (logger.isTraceEnabled()) { logger.trace("address = {} found a duplicate {}", address, describeID(duplID.bytes)); @@ -258,7 +262,7 @@ private synchronized void addToCache(final ByteArray holder, } if (instantAdd) { - addToCacheInMemory(holder, recordID); + reserve(holder); tx.addOperation(new AddDuplicateIDOperation(holder, recordID, false)); } else { // For a tx, it's important that the entry is not added to the cache until commit @@ -273,6 +277,14 @@ public void load(final Transaction tx, final byte[] duplID) { tx.addOperation(new AddDuplicateIDOperation(new ByteArray(duplID), tx.getID(), true)); } + private void reserve(final ByteArray holder) { + inFlight.merge(holder, 1, Integer::sum); + } + + private void release(final ByteArray holder) { + inFlight.computeIfPresent(holder, (id, count) -> count == 1 ? null : count - 1); + } + private synchronized void addToCacheInMemory(final ByteArray holder, final long recordID) { Objects.requireNonNull(holder, "holder must be not null"); if (recordID < 0) { @@ -394,6 +406,15 @@ private void process() { } } + @Override + public void beforeCommit(final Transaction tx) throws Exception { + // The send was routed: its id takes its slot now, as a send outside a transaction does + if (!afterCommit) { + process(); + release(holder); + } + } + @Override public void afterCommit(final Transaction tx) { if (afterCommit) { @@ -404,7 +425,7 @@ public void afterCommit(final Transaction tx) { @Override public void beforeRollback(Transaction tx) throws Exception { if (!afterCommit) { - deleteFromCache(holder, null); + release(holder); } } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/DuplicateDetectionTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/DuplicateDetectionTest.java index 291085aa03ad..a712e47175d0 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/DuplicateDetectionTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/DuplicateDetectionTest.java @@ -19,6 +19,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import static org.junit.jupiter.api.Assumptions.assumeTrue; @@ -30,6 +31,7 @@ import java.util.Arrays; import java.util.Collection; +import org.apache.activemq.artemis.api.core.ActiveMQAddressFullException; import org.apache.activemq.artemis.api.core.ActiveMQDuplicateIdException; import org.apache.activemq.artemis.api.core.ActiveMQException; import org.apache.activemq.artemis.api.core.Message; @@ -47,6 +49,7 @@ import org.apache.activemq.artemis.core.config.Configuration; import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl; import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy; import org.apache.activemq.artemis.core.settings.impl.AddressSettings; import org.apache.activemq.artemis.core.transaction.impl.XidImpl; import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter; @@ -55,6 +58,7 @@ import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; import org.apache.activemq.artemis.utils.RandomUtil; import org.apache.activemq.artemis.utils.UUIDGenerator; +import org.apache.activemq.artemis.utils.Wait; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.extension.ExtendWith; @@ -918,6 +922,64 @@ public void testRollbackThenSend() throws Exception { assertEquals(dupID1.toString(), message.getStringProperty("key")); } + /* + * A send refused after its duplicate check leaves the cache as it was: the id before it is still a duplicate + */ + @TestTemplate + public void testRefusedSendKeepsTheIdBeforeIt() throws Exception { + final SimpleString queueName = SimpleString.of("FullQueue"); + + server.getAddressSettingsRepository().addMatch(queueName.toString(), new AddressSettings().setIDCacheSize(1).setMaxSizeBytes(10000).setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL)); + + ClientSession session = sf.createSession(false, true, true); + + session.start(); + + session.createQueue(QueueConfiguration.of(queueName).setDurable(false)); + + ClientProducer producer = session.createProducer(queueName); + + producer.send(createLargeBody(session, "d1")); + + int sent = 1; + while (true) { + try { + producer.send(createLargeBody(session, null)); + sent++; + assertTrue(sent < 20, "the address never filled"); + } catch (ActiveMQAddressFullException e) { + break; + } + } + + assertThrows(ActiveMQAddressFullException.class, () -> producer.send(createLargeBody(session, "d2"))); + + ClientConsumer consumer = session.createConsumer(queueName); + for (int i = 0; i < sent; i++) { + ClientMessage message = consumer.receive(5000); + assertNotNull(message); + message.acknowledge(); + } + assertNull(consumer.receiveImmediate()); + // Sends the acknowledgements + consumer.close(); + Wait.assertEquals(0L, () -> server.getPagingManager().getPageStore(queueName).getAddressSize()); + + producer.send(createLargeBody(session, "d1")); + + consumer = session.createConsumer(queueName); + assertNull(consumer.receive(500)); + } + + private ClientMessage createLargeBody(final ClientSession session, final String dupID) { + ClientMessage message = session.createMessage(true); + message.getBodyBuffer().writeBytes(new byte[4000]); + if (dupID != null) { + message.putStringProperty(Message.HDR_DUPLICATE_DETECTION_ID, SimpleString.of(dupID)); + } + return message; + } + /* * Entire transaction should be rejected on duplicate detection * Even if not all entries have dupl id header