From be0a5b327fff6f2d84bd22a37ffea735c7306d5b Mon Sep 17 00:00:00 2001 From: grepjava Date: Sun, 8 Nov 2026 19:56:00 +0800 Subject: [PATCH] ARTEMIS-6360 A second send of a duplicate id in the same transaction is a duplicate --- .../core/postoffice/impl/PostOfficeImpl.java | 29 +++++++++++- .../TransactionPropertyIndexes.java | 2 + .../integration/DuplicateDetectionTest.java | 45 +++++++++++++++++++ 3 files changed, 75 insertions(+), 1 deletion(-) diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java index d73976318049..76e5eed1fd5e 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java @@ -1911,7 +1911,7 @@ private DuplicateCheckResult checkNotBridgeDuplicateID(final Message message, final byte[] duplicateIDBytes) throws Exception { assert duplicateIDBytes != null && Arrays.equals(message.getDuplicateIDBytes(), duplicateIDBytes); final DuplicateIDCache cache = getDuplicateIDCache(context.getAddress(message)); - final boolean isDuplicate = cache.contains(duplicateIDBytes); + final boolean isDuplicate = cache.contains(duplicateIDBytes) || sentInTransaction(context.getTransaction(), cache, duplicateIDBytes); if (rejectDuplicates && isDuplicate) { ActiveMQServerLogger.LOGGER.duplicateMessageDetected(message); if (context.getTransaction() != null) { @@ -1934,9 +1934,36 @@ private DuplicateCheckResult checkNotBridgeDuplicateID(final Message message, startedTX = false; } cache.addToCache(duplicateIDBytes, context.getTransaction(), startedTX); + if (!startedTX) { + // The id joins the cache only when the transaction commits: until then a second send of it in the same + // transaction is found here + sentInTransaction(context.getTransaction()).computeIfAbsent(cache, c -> new HashSet<>()).add(new ByteArray(duplicateIDBytes)); + } return startedTX ? DuplicateCheckResult.NoDuplicateStartedTX : DuplicateCheckResult.NoDuplicateNotStartedTX; } + /** + * The duplicate ids the transaction has sent, by cache. + */ + @SuppressWarnings("unchecked") + private static Map> sentInTransaction(final Transaction tx) { + Map> sent = (Map>) tx.getProperty(TransactionPropertyIndexes.DUPLICATE_IDS); + if (sent == null) { + sent = new HashMap<>(); + tx.putProperty(TransactionPropertyIndexes.DUPLICATE_IDS, sent); + } + return sent; + } + + @SuppressWarnings("unchecked") + private static boolean sentInTransaction(final Transaction tx, final DuplicateIDCache cache, final byte[] duplicateIDBytes) { + if (tx == null) { + return false; + } + final Map> sent = (Map>) tx.getProperty(TransactionPropertyIndexes.DUPLICATE_IDS); + return sent != null && sent.getOrDefault(cache, Collections.emptySet()).contains(new ByteArray(duplicateIDBytes)); + } + private DuplicateCheckResult checkBridgeDuplicateID(final Message message, final RoutingContext context, final byte[] bridgeDupBytes) throws Exception { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionPropertyIndexes.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionPropertyIndexes.java index ee80922647d8..3e1d910a3a34 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionPropertyIndexes.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionPropertyIndexes.java @@ -45,4 +45,6 @@ public class TransactionPropertyIndexes { public static final int MIRROR_DELIVERY_ASYNC = 13; public static final int ADD_MESSAGE = 14; + + public static final int DUPLICATE_IDS = 15; } 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..2fd889bd67bd 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; @@ -918,6 +919,50 @@ public void testRollbackThenSend() throws Exception { assertEquals(dupID1.toString(), message.getStringProperty("key")); } + /* + * A second send of an id in the same transaction is a duplicate, though the id joins the cache only at the commit + */ + @TestTemplate + public void testSameIdTwiceInOneTransaction() throws Exception { + ClientSession session = sf.createSession(false, false, false); + + session.start(); + + final SimpleString queueName = SimpleString.of("DuplicateDetectionTestQueue"); + + session.createQueue(QueueConfiguration.of(queueName).setDurable(false)); + + ClientProducer producer = session.createProducer(queueName); + + ClientConsumer consumer = session.createConsumer(queueName); + + SimpleString dupID = SimpleString.of("abcdefg"); + + for (int i = 0; i < 2; i++) { + ClientMessage message = createMessage(session, i); + message.putBytesProperty(Message.HDR_DUPLICATE_DETECTION_ID, dupID.getData()); + producer.send(message); + } + + assertThrows(ActiveMQDuplicateIdException.class, session::commit); + session.rollback(); + + assertNull(consumer.receiveImmediate()); + + // The id was not taken: a new transaction sends it + ClientMessage message = createMessage(session, 2); + message.putBytesProperty(Message.HDR_DUPLICATE_DETECTION_ID, dupID.getData()); + producer.send(message); + + session.commit(); + + message = consumer.receive(5000); + assertNotNull(message); + assertEquals(2, message.getObjectProperty(propKey)); + + assertNull(consumer.receiveImmediate()); + } + /* * Entire transaction should be rejected on duplicate detection * Even if not all entries have dupl id header