ARTEMIS-6146 Large Message inconsistencies through Divert and Mirror

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

@@ -215,6 +215,14 @@ default void clearInternalProperties() {
    default void clearAMQPProperties() {
    }
 
+   default boolean isDropped() {
+      return false;
+   }
+
+   default Message setDropped(boolean dropped) {
+      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 setDropped(boolean dropped) {
+      this.dropped = dropped;
+      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 setDropped(boolean dropped) {
+      this.dropped = dropped;
+      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.setDropped(true);
+
             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.setDropped(true);
+
             if (message.isLargeMessage()) {
                ((LargeServerMessage) message).deleteFile();
             }

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

@@ -1367,6 +1367,7 @@ private RoutingStatus maybeSendToDLA(final Message message,
          status = RoutingStatus.NO_BINDINGS;
          logger.debug("Message {} is not going anywhere as it didn't have a binding on address:{}", message, address);
          if (message.isLargeMessage()) {
+            message.setDropped(true);
             ((LargeServerMessage) message).deleteFile();
          }
       }
@@ -1700,11 +1701,19 @@ public void processRoute(final Message message,
          }
 
          if (store != null && storageManager.addToPage(store, message, context.getTransaction(), entry.getValue())) {
-            // We need to kick delivery so the Queues may check for the cursors case they are empty
-            schedulePageDelivery(tx, entry);
+            if (!message.isDropped()) {
+               // we schedule prefetch checks when messages are added
+               schedulePageDelivery(tx, entry);
+            }
             continue;
          }
 
+         if (message.isDropped()) {
+            // this should never happen
+            // adding defensive code just in case
+            throw new IllegalStateException("Paging returned false (NOT_PAGED) for a dropped message");
+         }
+
          final List<Queue> nonDurableQueues = entry.getValue().getNonDurableQueues();
          if (!nonDurableQueues.isEmpty()) {
             refs.ensureCapacity(nonDurableQueues.size());
@@ -1725,12 +1734,20 @@ public void processRoute(final Message message,
          }
       }
 
+      if (message.isDropped()) {
+         if (startedTX) {
+            // the transaction here should be empty
+            // calling rollback just for correctness
+            tx.rollback();
+         }
+         return;
+      }
+
       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) {

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

@@ -153,7 +153,7 @@ public void route(final Message message, final RoutingContext context) throws Ex
             copy = message;
          }
 
-         postOffice.route(copy, new RoutingContextImpl(context.getTransaction()).setReusable(false).setRoutingType(copy.getRoutingType()).setServerSession(context.getServerSession()), false);
+         postOffice.route(copy, new RoutingContextImpl(context.getTransaction()).setReusable(false).setMirrorOption(context.getMirrorOption()).setRoutingType(copy.getRoutingType()).setServerSession(context.getServerSession()), false);
       }
    }
 

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

@@ -3895,6 +3895,11 @@ public void errorProcessing(Consumer consumer, Throwable t, MessageReference ref
    }
 
    private boolean checkExpired(final MessageReference reference) {
+      if (isMirrorController()) {
+         // we don't expire through the MirrorSNF
+         return false;
+      }
+
       try {
          if (reference.getMessage().isExpired()) {
             logger.trace("Reference {} is expired", reference);

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

@@ -16,14 +16,47 @@
  */
 package org.apache.activemq.artemis.core.postoffice.impl;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
 import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.config.WildcardConfiguration;
+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.persistence.impl.journal.LargeServerMessageImpl;
+import org.apache.activemq.artemis.core.postoffice.QueueBinding;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.ComponentConfigurationRoutingType;
+import org.apache.activemq.artemis.core.server.QueueFactory;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.DivertImpl;
+import org.apache.activemq.artemis.core.server.impl.QueueImpl;
+import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl;
+import org.apache.activemq.artemis.core.server.management.ManagementService;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
+import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
 import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.core.settings.impl.HierarchicalObjectRepository;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
+import static org.junit.jupiter.api.Assertions.assertFalse;
 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.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.doAnswer;
+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 +338,178 @@ public void testPrecedencExpiryDelayOverMinExpiryDelay() {
 
       assertExpirationSetAsExpected(expectedExpirationLow, expectedExpirationHigh, actualExpirationSet);
    }
+
+   @Test
+   public void testProcessRouteDoesNotCallMirrorControllerWhenMessageDropped() throws Exception {
+
+      // Setup Members (mocks and real objects) to be used on PostOffice
+
+      StorageManager storageManager = mock(StorageManager.class);
+      AtomicLong sequence = new AtomicLong(1L);
+      when(storageManager.generateID()).thenAnswer(invocation -> sequence.incrementAndGet());
+
+      ActiveMQServer server = mock(ActiveMQServer.class);
+      PagingManager pagingManager = mock(PagingManager.class);
+      PagingStore pagingStore = mock(PagingStore.class);
+      MirrorController mirrorController = mock(MirrorController.class);
+      ManagementService managementService = mock(ManagementService.class);
+      QueueFactory queueFactory = mock(QueueFactory.class);
+      WildcardConfiguration wildcardConfiguration = new WildcardConfiguration();
+      HierarchicalRepository<AddressSettings> hierarchicalRepository = new HierarchicalObjectRepository<>();
+
+      PostOfficeImpl postOffice = new PostOfficeImpl(server, storageManager, pagingManager, queueFactory,
+                                                     managementService, 100, 100,
+                                                     wildcardConfiguration, -1, false, hierarchicalRepository).setMirrorControlSource(mirrorController);
+
+      SimpleString address = RandomUtil.randomUUIDSimpleString();
+
+      Message message = mock(Message.class);
+      final AtomicBoolean dropped = new AtomicBoolean();
+      when(message.getAddressSimpleString()).thenReturn(address);
+      when(message.hasScheduledDeliveryTime()).thenReturn(false);
+      when(message.isDurable()).thenReturn(false);
+      when(message.isDropped()).thenAnswer(invocation -> dropped.get());
+      when(message.setDropped(true)).thenAnswer(invocation -> {
+         dropped.set(true);
+         return message;
+      });
+
+      QueueImpl mockedQueue = Mockito.mock(QueueImpl.class);
+
+      RoutingContext context = new RoutingContextImpl(null);
+      context.addQueue(address, mockedQueue);
+
+      when(pagingStore.getAddress()).thenReturn(address);
+      when(pagingManager.getPageStore(address)).thenReturn(pagingStore);
+
+      when(storageManager.addToPage(eq(pagingStore), any(), any(), any()))
+         .thenAnswer(invocation -> {
+            // Simulate message.drop() being called inside addToPage
+            message.setDropped(true);
+            return true;
+         });
+
+      postOffice.processRoute(message, context, false);
+
+      assertTrue(dropped.get());
+
+      verify(mirrorController, never()).sendMessage(any(), any(), any());
+
+      verify(mockedQueue, never()).refUp(any());
+   }
+
+   // test that when mirror is disabled, the divert routing countext will follow along with the initial configuration
+   // for example, the expiration disabled mirroring, and if a 'divert' is configured on the expiration,
+   // that divert operation should also have mirror disabled
+   @Test
+   public void testMirrorDisablePropagation() throws Exception {
+      // Setup Members (mocks and real objects) to be used on PostOffice
+
+      StorageManager storageManager = mock(StorageManager.class);
+      AtomicLong sequence = new AtomicLong(1L);
+      when(storageManager.generateID()).thenAnswer(invocation -> sequence.incrementAndGet());
+
+      ActiveMQServer server = mock(ActiveMQServer.class);
+      PagingManager pagingManager = mock(PagingManager.class);
+      PagingStore pagingStore = mock(PagingStore.class);
+      MirrorController mirrorController = mock(MirrorController.class);
+      ManagementService managementService = mock(ManagementService.class);
+      QueueFactory queueFactory = mock(QueueFactory.class);
+      WildcardConfiguration wildcardConfiguration = new WildcardConfiguration().setRoutingEnabled(false);
+      HierarchicalRepository<AddressSettings> hierarchicalRepository = new HierarchicalObjectRepository<>();
+
+      PostOfficeImpl postOffice = new PostOfficeImpl(server, storageManager, pagingManager, queueFactory,
+                                                     managementService, 100, 100,
+                                                     wildcardConfiguration, -1, false, hierarchicalRepository);
+
+
+      postOffice.setMirrorControlSource(mirrorController);
+
+      SimpleString address = SimpleString.of("sourceAddress");
+      SimpleString forwardAddress = SimpleString.of("destinationForDivert");
+      DivertBinding divertBinding = new DivertBinding(sequence.incrementAndGet(), address, new DivertImpl(address, address, forwardAddress, forwardAddress, true, null, null, postOffice, storageManager, ComponentConfigurationRoutingType.ANYCAST));
+
+      QueueImpl targetDivertQueue = mock(QueueImpl.class);
+      when(targetDivertQueue.getName()).thenReturn(forwardAddress);
+      doAnswer(f -> {
+         RoutingContext context = f.getArgument(1);
+         // simulating the routing of a real queue
+         context.addQueue(forwardAddress, targetDivertQueue);
+         return null;
+      }).when(targetDivertQueue).route(any(), any());
+
+      when(targetDivertQueue.getAddress()).thenReturn(forwardAddress);
+
+      QueueBinding queueBinding = new LocalQueueBinding(forwardAddress, targetDivertQueue, RandomUtil.randomUUIDSimpleString());
+
+      postOffice.addBinding(queueBinding);
+      postOffice.addBinding(divertBinding);
+
+      LargeServerMessageImpl divertMessage = mock(LargeServerMessageImpl.class);
+      when(divertMessage.getAddressSimpleString()).thenReturn(forwardAddress);
+      when(divertMessage.hasScheduledDeliveryTime()).thenReturn(false);
+      when(divertMessage.isDurable()).thenReturn(false);
+      when(divertMessage.isLargeMessage()).thenReturn(true);
+      {
+         final AtomicBoolean dropped = new AtomicBoolean();
+         when(divertMessage.isDropped()).thenAnswer(invocation -> dropped.get());
+         when(divertMessage.setDropped(true)).thenAnswer(invocation -> {
+            dropped.set(true);
+            return divertMessage;
+         });
+      }
+
+      Mockito.doAnswer(f -> {
+         Assertions.fail("not supposed to delete the target message");
+         return null;
+      }).when(divertMessage).deleteFile();
+
+      LargeServerMessageImpl message = mock(LargeServerMessageImpl.class);
+      when(message.getAddressSimpleString()).thenReturn(address);
+      when(message.hasScheduledDeliveryTime()).thenReturn(false);
+      when(message.isDurable()).thenReturn(false);
+      {
+         final AtomicBoolean dropped = new AtomicBoolean();
+         when(message.isDropped()).thenAnswer(invocation -> dropped.get());
+         when(message.setDropped(true)).thenAnswer(invocation -> {
+            dropped.set(true);
+            return message;
+         });
+      }
+
+      when(message.isLargeMessage()).thenReturn(true);
+
+      AtomicBoolean messageDeleted = new AtomicBoolean();
+      when(message.copy(anyLong())).thenAnswer(f -> {
+         Assertions.assertFalse(messageDeleted.get()); // it would be an issue if we copied a deleted message
+         return divertMessage;
+      });
+      Mockito.doAnswer(f -> {
+         messageDeleted.set(true);
+         return null;
+      }).when(message).deleteFile();
+
+      when(pagingStore.getAddress()).thenReturn(address);
+      when(pagingManager.getPageStore(address)).thenReturn(pagingStore);
+
+      when(storageManager.addToPage(eq(pagingStore), any(), any(), any()))
+         .thenAnswer(invocation -> {
+            return false; // Returns true indicating the message was added to page
+         });
+
+      assertFalse(messageDeleted.get());
+      assertFalse(message.isDropped());
+
+      RoutingContext routingContext = new RoutingContextImpl(null).setMirrorOption(RoutingContext.MirrorOption.disabled);
+      postOffice.route(message, routingContext, false);
+
+      assertTrue(messageDeleted.get()); // the original message should have been deleted
+      assertTrue(message.isDropped()); // the original message should have been dropped
+      assertFalse(divertMessage.isDropped()); // the target diverted message should not be dropped
+
+      // We disabled mirror on this routing, so if mirrorController was called, it means that the divert is reverting the setting
+      verify(mirrorController, never()).sendMessage(any(), any(), any());
+      // refUp should be called on the target divert
+      verify(targetDivertQueue, atLeast(1)).refUp(any());
+   }
 }