diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java index fe647e48387..8d03a857a38 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java +++ b/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. */ diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java index 40e750216d1..99ca6d29f95 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java +++ b/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; diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java index a989fac6827..253943aa202 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java +++ b/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 diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java index 39b9c3bec46..1ba73dcffa0 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java +++ b/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(); } 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 c8de8955a04..6ee9a03a1a9 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 @@ -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 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) { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java index 526971d54b0..30e79bf4251 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java +++ b/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); } } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java index 8632fc3a844..04a47ad3aa4 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java +++ b/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); diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImplTest.java index 3730a1ee7e0..446f03e15c5 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImplTest.java +++ b/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 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 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()); + } } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPMirrorLargeMessageDivertDroppedTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPMirrorLargeMessageDivertDroppedTest.java new file mode 100644 index 00000000000..4c6b0080d02 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPMirrorLargeMessageDivertDroppedTest.java @@ -0,0 +1,354 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.tests.integration.amqp.connect; + +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; +import java.lang.invoke.MethodHandles; +import java.util.HashMap; +import java.util.Map; + +import org.apache.activemq.artemis.api.core.QueueConfiguration; +import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.api.core.TransportConfiguration; +import org.apache.activemq.artemis.core.config.DivertConfiguration; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPMirrorBrokerConnectionElement; +import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.Queue; +import org.apache.activemq.artemis.core.server.impl.QueueImplTestAccessor; +import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy; +import org.apache.activemq.artemis.core.settings.impl.AddressSettings; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.tests.util.CFUtil; +import org.apache.activemq.artemis.tests.util.Wait; +import org.apache.activemq.artemis.utils.RandomUtil; +import org.jspecify.annotations.NonNull; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +public class AMQPMirrorLargeMessageDivertDroppedTest extends ActiveMQTestBase { + + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + // notice I'm not placing this as static as I prefer such a large buffer to be referenced from the test instance + // which I know Garbage Collection happens much earlier than the class release. + private final String LARGE_BUFFER = RandomUtil.randomAlphaNumericString(300 * 1024); + + private static final long DROP_LIMIT = 5; + private static final long NUMBER_OF_MESSAGES = 10; + private static final int PORT_SERVER_A = 5671; + private static final int PORT_SERVER_B = 6671; + private static final String URI_SERVER_A = "tcp://localhost:" + PORT_SERVER_A; + private static final String URI_SERVER_B = "tcp://localhost:" + PORT_SERVER_B; + + // Basic queue is a simple queue, no diverts. We send stuff to validate mirror isn't interrupted and everything works + private static final String BASIC_QUEUE = AMQPMirrorLargeMessageDivertDroppedTest.class.getName() + "_BASIC"; + + // Expiry has a 'divert' towards ExpiryDivert + private static final String EXPIRY_QUEUE = AMQPMirrorLargeMessageDivertDroppedTest.class.getName() + "_ExpiryOut"; + private static final String EXPIRY_DIVERT = AMQPMirrorLargeMessageDivertDroppedTest.class.getName() + "_ExpiryDiverted"; + + // RegularQueue has a 'divert' towards RegularDivert + private static final String REGULAR_QUEUE = AMQPMirrorLargeMessageDivertDroppedTest.class.getName() + "_RegularQueue"; + private static final String REGULAR_QUEUE_DIVERT = AMQPMirrorLargeMessageDivertDroppedTest.class.getName() + "RegularQueueDivert"; + + ActiveMQServer serverA; + ActiveMQServer serverB; + + protected TransportConfiguration newAcceptorConfig(int port, String name) { + Map params = new HashMap<>(); + params.put(TransportConstants.PORT_PROP_NAME, String.valueOf(port)); + params.put(TransportConstants.PROTOCOLS_PROP_NAME, "AMQP,CORE"); + Map amqpParams = new HashMap<>(); + TransportConfiguration tc = new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params, name, amqpParams); + return tc; + } + + protected ActiveMQServer createServer(int port, String brokerName) throws Exception { + + final ActiveMQServer server = this.createServer(true, true); + + server.getConfiguration().getAcceptorConfigurations().clear(); + server.getConfiguration().getAcceptorConfigurations().add(newAcceptorConfig(port, "netty-acceptor")); + server.getConfiguration().setName(brokerName); + server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port); + server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port); + server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port); + server.getConfiguration().setLargeMessagesDirectory(server.getConfiguration().getLargeMessagesDirectory() + port); + server.getConfiguration().setMessageExpiryScanPeriod(5); + + server.getConfiguration().addAddressSetting("#", new AddressSettings().setExpiryAddress(SimpleString.of(EXPIRY_QUEUE))); + + server.getConfiguration().addQueueConfiguration(QueueConfiguration.of(EXPIRY_QUEUE).setRoutingType(RoutingType.ANYCAST)); + server.getConfiguration().addQueueConfiguration(QueueConfiguration.of(EXPIRY_DIVERT).setRoutingType(RoutingType.ANYCAST)); + + server.getConfiguration().addQueueConfiguration(QueueConfiguration.of(REGULAR_QUEUE).setRoutingType(RoutingType.ANYCAST)); + server.getConfiguration().addQueueConfiguration(QueueConfiguration.of(REGULAR_QUEUE_DIVERT).setRoutingType(RoutingType.ANYCAST)); + + server.getConfiguration().addQueueConfiguration(QueueConfiguration.of(BASIC_QUEUE).setRoutingType(RoutingType.ANYCAST)); + + { + DivertConfiguration divertConfiguration = new DivertConfiguration().setName("divertExpiry").setAddress(EXPIRY_QUEUE).setForwardingAddress(EXPIRY_DIVERT).setExclusive(true); + server.getConfiguration().addDivertConfiguration(divertConfiguration); + } + + { + DivertConfiguration divertConfiguration = new DivertConfiguration().setName("divertRegular").setAddress(REGULAR_QUEUE).setForwardingAddress(REGULAR_QUEUE_DIVERT).setExclusive(true); + server.getConfiguration().addDivertConfiguration(divertConfiguration); + } + + return server; + } + + @Test + public void testDropThroughExpiryAMQP() throws Exception { + startServers(); + internalTestDropThroughExpiry("AMQP"); + } + + @Test + public void testDropThroughExpiryCORE() throws Exception { + startServers(); + internalTestDropThroughExpiry("CORE"); + } + + @Test + public void testDropThroughRegularAMQP() throws Exception { + startServers(); + internalTestDropThroughRegular("AMQP"); + } + + @Test + public void testDropThroughRegularCORE() throws Exception { + startServers(); + internalTestDropThroughRegular("CORE"); + } + + private void internalTestDropThroughExpiry(String protocol) throws Exception { + String queueName = getTestMethodName() + "_" + RandomUtil.randomUUIDString(); + serverA.createQueue(QueueConfiguration.of(queueName).setName(queueName).setRoutingType(RoutingType.ANYCAST)); + + Queue expiryA = serverA.locateQueue(EXPIRY_QUEUE); + assertNotNull(expiryA); + + Queue expiryDiverted = serverA.locateQueue(EXPIRY_DIVERT); + assertNotNull(expiryDiverted); + assertEquals(0, expiryDiverted.getMessageCount()); + + Queue expiryDivertedB = serverB.locateQueue(EXPIRY_DIVERT); + assertNotNull(expiryDivertedB); + assertEquals(0, expiryDivertedB.getMessageCount()); + + Queue snfQueue = serverA.locateQueue(QueueConfiguration.MIRROR_ADDRESS + "_" + getTestMethodName()); + assertNotNull(snfQueue); + Wait.assertEquals(0L, snfQueue::getMessageCount, 5000, 100); + + ConnectionFactory factoryA = CFUtil.createConnectionFactory(protocol, URI_SERVER_A); + ConnectionFactory factoryB = CFUtil.createConnectionFactory(protocol, URI_SERVER_B); + + try (Connection connection = factoryA.createConnection()) { + try (Session session = connection.createSession(true, Session.SESSION_TRANSACTED)) { + try (MessageProducer producer = session.createProducer(session.createQueue(queueName))) { + producer.setTimeToLive(100); + + Wait.assertEquals(0L, expiryDiverted::getMessageCount, 5000, 100); + + for (int i = 0; i < NUMBER_OF_MESSAGES; i++) { + producer.send(session.createTextMessage(generateTextBody(i))); + } + session.commit(); + Wait.assertEquals(DROP_LIMIT, expiryDiverted::getMessageCount, 5000, 100); + } + } + } + + Wait.assertEquals(DROP_LIMIT, expiryDiverted::getMessageCount, 5000, 100); + + // Expiry happens on target, we should have all message expiring on target + // I disabled DROP on target to valdiate this + Wait.assertEquals(NUMBER_OF_MESSAGES, expiryDivertedB::getMessageCount, 5000, 100); + + Wait.assertEquals(0L, snfQueue::getMessageCount, 5000, 100); + + // Firs we consume from B, as there's no DROP limit there, we validate the messages + consumeMessages(factoryB, EXPIRY_DIVERT, NUMBER_OF_MESSAGES); + consumeMessages(factoryA, EXPIRY_DIVERT, DROP_LIMIT); + + Wait.assertEquals(0L, expiryDiverted::getMessageCount); + Wait.assertEquals(0L, expiryDivertedB::getMessageCount); + + assertEquals(0, QueueImplTestAccessor.getQueueMemorySize(expiryDiverted)); + assertEquals(0, QueueImplTestAccessor.getQueueMemorySize(expiryDivertedB)); + + // This is to validate Mirror was able to finish the load + // The issue the test was reproducing was an issue where mirror would disconnect and never complete execution + Wait.assertEquals(0L, snfQueue::getMessageCount, 5000, 100); + + // validate the mirror still operational + validateBasicQueue(protocol); + } + + private void consumeMessages(ConnectionFactory factory, String queueName, long numberOfMessages) throws JMSException { + try (Connection connection = factory.createConnection()) { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageConsumer consumer = session.createConsumer(session.createQueue(queueName)); + connection.start(); + for (int i = 0; i < numberOfMessages; i++) { + TextMessage message = (TextMessage) consumer.receive(5000); + assertNotNull(message); + assertEquals(generateTextBody(i), message.getText()); + } + assertNull(consumer.receiveNoWait()); + } + } + + private @NonNull String generateTextBody(int i) { + return "hello" + i + "_" + LARGE_BUFFER; + } + + // to make sure basic operation still works fine on the mirror + private void validateBasicQueue(String protocol) throws Exception { + ConnectionFactory cfA = CFUtil.createConnectionFactory(protocol, URI_SERVER_A); + + { + try (Connection connection = cfA.createConnection(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(session.createQueue(BASIC_QUEUE))) { + for (int i = 0; i < NUMBER_OF_MESSAGES; i++) { + producer.send(session.createTextMessage("hello" + i)); + } + } + } + + Queue basicA = serverA.locateQueue(BASIC_QUEUE); + assertNotNull(basicA); + Wait.assertEquals(NUMBER_OF_MESSAGES, basicA::getMessageCount, 5000, 100); + + Queue basicB = serverB.locateQueue(BASIC_QUEUE); + assertNotNull(basicB); + Wait.assertEquals(NUMBER_OF_MESSAGES, basicB::getMessageCount, 5000, 100); + + { + try (Connection connectionA = cfA.createConnection(); + Session sessionA = connectionA.createSession(true, Session.SESSION_TRANSACTED); + MessageConsumer consumerA = sessionA.createConsumer(sessionA.createQueue(BASIC_QUEUE))) { + connectionA.start(); + for (int i = 0; i < NUMBER_OF_MESSAGES; i++) { + TextMessage message = (TextMessage) consumerA.receive(1000); + assertNotNull(message); + assertEquals("hello" + i, message.getText()); + } + sessionA.commit(); + } + } + + Wait.assertEquals(0L, basicA::getMessageCount, 5000, 100); + Wait.assertEquals(0L, basicB::getMessageCount, 5000, 100); + + } + + private void internalTestDropThroughRegular(String protocol) throws Exception { + Queue regularQueueA = serverA.locateQueue(REGULAR_QUEUE); + assertNotNull(regularQueueA); + + Queue regularQueueB = serverB.locateQueue(REGULAR_QUEUE); + assertNotNull(regularQueueB); + + Queue regularDivertedA = serverA.locateQueue(REGULAR_QUEUE_DIVERT); + assertNotNull(regularDivertedA); + assertEquals(0, regularDivertedA.getMessageCount()); + + Queue regularDivertedB = serverB.locateQueue(REGULAR_QUEUE_DIVERT); + assertNotNull(regularDivertedB); + assertEquals(0, regularDivertedB.getMessageCount()); + + Queue snfQueue = serverA.locateQueue(QueueConfiguration.MIRROR_ADDRESS + "_" + getTestMethodName()); + assertNotNull(snfQueue); + Wait.assertEquals(0L, snfQueue::getMessageCount, 5000, 100); + + ConnectionFactory factoryA = CFUtil.createConnectionFactory(protocol, URI_SERVER_A); + ConnectionFactory factoryB = CFUtil.createConnectionFactory(protocol, URI_SERVER_B); + + try (Connection connection = factoryA.createConnection()) { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(session.createQueue(REGULAR_QUEUE)); + + for (int i = 0; i < NUMBER_OF_MESSAGES; i++) { + producer.send(session.createTextMessage(generateTextBody(i))); + } + } + + Wait.assertEquals(DROP_LIMIT, regularDivertedA::getMessageCount, 5000, 100); + consumeMessages(factoryA, REGULAR_QUEUE_DIVERT, DROP_LIMIT); + + Wait.assertEquals(0L, regularDivertedA::getMessageCount, 5000, 100); + + assertEquals(0, regularQueueA.getMessageCount()); + // making sure the sizes match + assertEquals(0L, QueueImplTestAccessor.getQueueMemorySize(regularQueueA)); + assertEquals(0, regularDivertedA.getMessageCount()); + assertEquals(0L, QueueImplTestAccessor.getQueueMemorySize(regularDivertedA)); + + Wait.assertEquals(0L, regularDivertedB::getMessageCount, 5000, 100); + // making sure the sizes match + assertEquals(0L, QueueImplTestAccessor.getQueueMemorySize(regularQueueB)); + assertEquals(0, regularDivertedB.getMessageCount()); + assertEquals(0L, QueueImplTestAccessor.getQueueMemorySize(regularDivertedB)); + + // This is to validate Mirror was able to finish the load + // The issue the test was reproducing was an issue where mirror would disconnect and never complete execution + Wait.assertEquals(0L, snfQueue::getMessageCount, 5000, 100); + assertEquals(0L, QueueImplTestAccessor.getQueueMemorySize(snfQueue)); + + // no messages should be consumed on ServerB + consumeMessages(factoryB, REGULAR_QUEUE_DIVERT, 0); + + // validate the mirror still operational + validateBasicQueue(protocol); + } + + private void startServers() throws Exception { + serverB = createServer(PORT_SERVER_B, getTestMethodName() + "_B"); + serverB.start(); + + serverA = createServer(PORT_SERVER_A, getTestMethodName() + "_A"); + // only setting drop on the main servers. This is to validate mirror semantics between the two servers + serverA.getConfiguration().addAddressSetting(REGULAR_QUEUE_DIVERT.toString(), new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.DROP).setMaxSizeMessages(DROP_LIMIT)); + serverA.getConfiguration().addAddressSetting(EXPIRY_DIVERT.toString(), new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.DROP).setMaxSizeMessages(DROP_LIMIT)); + AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration(getTestMethodName(), URI_SERVER_B).setReconnectAttempts(5).setRetryInterval(500); + AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setType(AMQPBrokerConnectionAddressType.MIRROR).setDurable(true).setMessageAcknowledgements(true); + amqpConnection.addElement(replica); + serverA.getConfiguration().addAMQPConnection(amqpConnection); + serverA.setIdentity(getTestMethodName() + "_A"); + serverA.start(); + } + +} \ No newline at end of file