ARTEMIS-6360 A second send of a duplicate id in the same transaction is a duplicate

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<DuplicateIDCache, Set<ByteArray>> sentInTransaction(final Transaction tx) {
+      Map<DuplicateIDCache, Set<ByteArray>> sent = (Map<DuplicateIDCache, Set<ByteArray>>) 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<DuplicateIDCache, Set<ByteArray>> sent = (Map<DuplicateIDCache, Set<ByteArray>>) 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 {

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;
 }

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