ARTEMIS-6146 Divert and Mirror causing issues after dropped messages

Assisted-by Claude on implementing PostOfficeImplTest's mock test

Author: clebertsuconic

artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java

@@ -215,6 +215,15 @@ default void clearInternalProperties() {
    default void clearAMQPProperties() {
    }
 
+   default boolean isDropped() {
+      return false;
+   }
+
+   // Marking the message as dropped
+   default Message drop() {
+      return this;
+   }
+
    /**
     * Search for the existence of the property: an implementor can save the message to be decoded, if possible.
     */

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java

@@ -117,6 +117,8 @@ public class CoreMessage extends RefCountMessage implements ICoreMessage {
 
    private volatile Object owner;
 
+   private boolean dropped;
+
    public CoreMessage(final CoreMessageObjectPools coreMessageObjectPools) {
       this.coreMessageObjectPools = coreMessageObjectPools;
    }
@@ -125,6 +127,17 @@ public CoreMessage() {
       this.coreMessageObjectPools = null;
    }
 
+   @Override
+   public CoreMessage drop() {
+      this.dropped = true;
+      return this;
+   }
+
+   @Override
+   public boolean isDropped() {
+      return dropped;
+   }
+
    @Override
    public void setPaged() {
       this.paged = true;

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java

@@ -211,6 +211,7 @@ private static void checkCode(int code) {
    protected long scheduledTime = -1;
    protected byte priority = DEFAULT_MESSAGE_PRIORITY;
 
+   protected boolean dropped;
    protected boolean isPaged;
    protected volatile boolean routed = false;
 
@@ -219,6 +220,17 @@ public void routed() {
       this.routed = true;
    }
 
+   @Override
+   public boolean isDropped() {
+      return dropped;
+   }
+
+   @Override
+   public AMQPMessage drop() {
+      this.dropped = true;
+      return this;
+   }
+
    // The Proton based AMQP message section that are retained in memory, these are the
    // mutable portions of the Message as the broker sees it, although AMQP defines that
    // the Properties and ApplicationProperties are immutable so care should be taken

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java

@@ -1438,6 +1438,8 @@ public int page(Message message,
 
       if (diskFullMessagePolicy == DiskFullMessagePolicy.DROP || diskFullMessagePolicy == DiskFullMessagePolicy.FAIL) {
          if (diskFull) {
+            message.drop();
+
             if (message.isLargeMessage()) {
                ((LargeServerMessage) message).deleteFile();
             }
@@ -1460,6 +1462,8 @@ public int page(Message message,
 
       if (addressFullMessagePolicy == AddressFullMessagePolicy.DROP || addressFullMessagePolicy == AddressFullMessagePolicy.FAIL) {
          if (full) {
+            message.drop();
+
             if (message.isLargeMessage()) {
                ((LargeServerMessage) message).deleteFile();
             }

artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java

@@ -1725,30 +1725,31 @@ public void processRoute(final Message message,
          }
       }
 
-      if (mirrorControllerSource != null && !context.isMirrorDisabled()) {
-         // we check for isMirrorDisabled as to avoid recursive loop from there
-         mirrorControllerSource.sendMessage(tx, message, context);
-      }
-
+      if (!message.isDropped()) {
+         if (mirrorControllerSource != null && !context.isMirrorDisabled()) {
+            // we check for isMirrorDisabled as to avoid recursive loop from there
+            mirrorControllerSource.sendMessage(tx, message, context);
+         }
 
-      if (tx != null) {
-         tx.addOperation(new AddOperation(refs));
-      } else if (!containsDurables) {
-         processReferences(refs, direct);
-      } else {
-         // This will use the same thread if there are no pending operations
-         // avoiding a context switch on this case
-         storageManager.afterCompleteOperations(new IOCallback() {
-            @Override
-            public void onError(final int errorCode, final String errorMessage) {
-               ActiveMQServerLogger.LOGGER.ioErrorAddingReferences(errorCode, errorMessage);
-            }
+         if (tx != null) {
+            tx.addOperation(new AddOperation(refs));
+         } else if (!containsDurables) {
+            processReferences(refs, direct);
+         } else {
+            // This will use the same thread if there are no pending operations
+            // avoiding a context switch on this case
+            storageManager.afterCompleteOperations(new IOCallback() {
+               @Override
+               public void onError(final int errorCode, final String errorMessage) {
+                  ActiveMQServerLogger.LOGGER.ioErrorAddingReferences(errorCode, errorMessage);
+               }
 
-            @Override
-            public void done() {
-               processReferences(refs, direct);
-            }
-         });
+               @Override
+               public void done() {
+                  processReferences(refs, direct);
+               }
+            });
+         }
       }
 
       if (startedTX) {

artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImplTest.java

@@ -16,14 +16,33 @@
  */
 package org.apache.activemq.artemis.core.postoffice.impl;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.paging.PagingManager;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RouteContextList;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
 import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.core.transaction.Transaction;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class PostOfficeImplTest {
 
@@ -305,4 +324,101 @@ public void testPrecedencExpiryDelayOverMinExpiryDelay() {
 
       assertExpirationSetAsExpected(expectedExpirationLow, expectedExpirationHigh, actualExpirationSet);
    }
+
+   /**
+    * Tests that mirrorController.sendMessage is NOT called when a message is dropped
+    * during storageManager.addToPage() call in processRoute().
+    * This test verifies the fix for ARTEMIS-6146 where dropped messages should not be mirrored.
+    *
+    * WITHOUT the fix: This test will FAIL because mirrorController.sendMessage will be called
+    * WITH the fix: This test will PASS because mirrorController.sendMessage will NOT be called
+    */
+   @Test
+   public void testProcessRouteDoesNotCallMirrorControllerWhenMessageDropped() throws Exception {
+      // Setup mocks
+      StorageManager storageManager = mock(StorageManager.class);
+      PagingManager pagingManager = mock(PagingManager.class);
+      PagingStore pagingStore = mock(PagingStore.class);
+      MirrorController mirrorController = mock(MirrorController.class);
+
+      // Create a spy of PostOfficeImpl to call the real processRoute method
+      PostOfficeImpl postOffice = Mockito.mock(PostOfficeImpl.class);
+
+      // Make processRoute call the real implementation
+      Mockito.doCallRealMethod().when(postOffice).processRoute(any(Message.class), any(RoutingContext.class), Mockito.anyBoolean());
+
+      // Mock the getMirrorControlSource to return our mock controller
+      when(postOffice.getMirrorControlSource()).thenReturn(mirrorController);
+
+      SimpleString address = SimpleString.of("testAddress");
+
+      // Create a message with dropped tracking
+      Message message = mock(Message.class);
+      final boolean[] dropped = {false};
+
+      when(message.getAddressSimpleString()).thenReturn(address);
+      when(message.hasScheduledDeliveryTime()).thenReturn(false);
+      when(message.isDurable()).thenReturn(false);
+      when(message.isDropped()).thenAnswer(invocation -> dropped[0]);
+      when(message.drop()).thenAnswer(invocation -> {
+         dropped[0] = true;
+         return message;
+      });
+
+      // Create route context
+      RoutingContext context = mock(RoutingContext.class);
+      RouteContextList routeContextList = mock(RouteContextList.class);
+      Map<SimpleString, RouteContextList> contextListing = new HashMap<>();
+      contextListing.put(address, routeContextList);
+
+      when(context.getContexListing()).thenReturn(contextListing);
+      when(context.getTransaction()).thenReturn(null);
+      when(context.isMirrorDisabled()).thenReturn(false);
+
+      // Configure route context list to return empty lists (message was paged)
+      when(routeContextList.getDurableQueues()).thenReturn(List.of());
+      when(routeContextList.getNonDurableQueues()).thenReturn(List.of());
+
+      // Configure paging store
+      when(pagingStore.getAddress()).thenReturn(address);
+      when(pagingManager.getPageStore(address)).thenReturn(pagingStore);
+
+      // Critical: When addToPage is called, it will drop the message
+      when(storageManager.addToPage(eq(pagingStore), eq(message), any(), eq(routeContextList)))
+         .thenAnswer(invocation -> {
+            // Simulate message.drop() being called inside addToPage
+            message.drop();
+            return true; // Returns true indicating the message was added to page
+         });
+
+      // Mock transaction creation for mirror controller
+      Transaction mockTx = mock(Transaction.class);
+      when(mockTx.setAsync(true)).thenReturn(mockTx);
+      when(storageManager.generateID()).thenReturn(1L);
+
+      // We need to inject the dependencies into the PostOfficeImpl mock
+      // Since we can't easily do that, we'll use reflection to set the fields
+      java.lang.reflect.Field storageManagerField = PostOfficeImpl.class.getDeclaredField("storageManager");
+      storageManagerField.setAccessible(true);
+      storageManagerField.set(postOffice, storageManager);
+
+      java.lang.reflect.Field pagingManagerField = PostOfficeImpl.class.getDeclaredField("pagingManager");
+      pagingManagerField.setAccessible(true);
+      pagingManagerField.set(postOffice, pagingManager);
+
+      java.lang.reflect.Field mirrorControllerField = PostOfficeImpl.class.getDeclaredField("mirrorControllerSource");
+      mirrorControllerField.setAccessible(true);
+      mirrorControllerField.set(postOffice, mirrorController);
+
+      // NOW call the REAL processRoute method
+      postOffice.processRoute(message, context, false);
+
+      // Verify the message was dropped
+      assertTrue(message.isDropped(), "Message should be marked as dropped after addToPage");
+
+      // This is the KEY assertion:
+      // WITHOUT the fix (isDropped check missing): mirrorController.sendMessage WILL be called -> TEST FAILS
+      // WITH the fix (isDropped check present): mirrorController.sendMessage will NOT be called -> TEST PASSES
+      verify(mirrorController, never()).sendMessage(any(), any(), any());
+   }
 }