From b4e2884f14ac467f58cb1a85479ece1cbe36df03 Mon Sep 17 00:00:00 2001 From: jbertram Date: Mon, 12 Jan 2015 10:16:00 -0600 Subject: [PATCH 1/2] Allow snapshots to fix the build after removing parent root --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index a84af9889f..4f3f440150 100644 --- a/pom.xml +++ b/pom.xml @@ -414,7 +414,7 @@ - false + true never From ec9ff4f55edf6e865f574c7be6cefa18db5dfeaa Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Wed, 7 Jan 2015 14:56:31 -0500 Subject: [PATCH 2/2] ACTIVEMQ6-68 Improving Scale down routine and fixing test https://issues.apache.org/jira/browse/ACTIVEMQ6-68 The logic on transferring method is a bit complex, where I found a better way to calculate the targetIDs. This will also fix a few ScaleDownTests I also added some extra tests using ScaleDownDirect --- .../activemq/utils/LinkedListIterator.java | 2 +- .../cursor/impl/PageSubscriptionImpl.java | 18 +- .../core/postoffice/AddressManager.java | 3 + .../activemq/core/postoffice/PostOffice.java | 3 + .../core/postoffice/impl/PostOfficeImpl.java | 5 + .../postoffice/impl/SimpleAddressManager.java | 11 + .../apache/activemq/core/server/Queue.java | 9 +- .../impl/BackupRecoveryJournalLoader.java | 2 +- .../core/server/impl/LiveOnlyActivation.java | 3 +- .../activemq/core/server/impl/QueueImpl.java | 41 +- .../core/server/impl/ScaleDownHandler.java | 631 ++++++++++-------- .../impl/ScheduledDeliveryHandlerTest.java | 4 +- .../activemq/tests/util/ServiceTestBase.java | 2 +- .../integration/client/HangConsumerTest.java | 5 +- .../client/InterruptedLargeMessageTest.java | 4 +- .../cluster/distribution/ClusterTestBase.java | 6 +- .../server/ScaleDownDirectTest.java | 319 +++++++++ .../integration/server/ScaleDownTest.java | 66 +- .../unit/core/postoffice/impl/FakeQueue.java | 4 +- .../server/impl/fakes/FakePostOffice.java | 7 + 20 files changed, 819 insertions(+), 326 deletions(-) create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownDirectTest.java diff --git a/activemq-core-client/src/main/java/org/apache/activemq/utils/LinkedListIterator.java b/activemq-core-client/src/main/java/org/apache/activemq/utils/LinkedListIterator.java index 4320dab31a..642dc80eff 100644 --- a/activemq-core-client/src/main/java/org/apache/activemq/utils/LinkedListIterator.java +++ b/activemq-core-client/src/main/java/org/apache/activemq/utils/LinkedListIterator.java @@ -28,7 +28,7 @@ import java.util.Iterator; * * */ -public interface LinkedListIterator extends Iterator +public interface LinkedListIterator extends Iterator, AutoCloseable { void repeat(); diff --git a/activemq-server/src/main/java/org/apache/activemq/core/paging/cursor/impl/PageSubscriptionImpl.java b/activemq-server/src/main/java/org/apache/activemq/core/paging/cursor/impl/PageSubscriptionImpl.java index 6482795668..e6dfc568a8 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/paging/cursor/impl/PageSubscriptionImpl.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/paging/cursor/impl/PageSubscriptionImpl.java @@ -1247,6 +1247,8 @@ final class PageSubscriptionImpl implements PageSubscription private volatile boolean isredelivery = false; + private PagedReference currentDelivery = null; + private volatile PagedReference lastRedelivery = null; // We only store the position for redeliveries. They will be read from the SoftCache again during delivery. @@ -1298,9 +1300,9 @@ final class PageSubscriptionImpl implements PageSubscription if (cachedNext != null) { - PagedReference retPos = cachedNext; + currentDelivery = cachedNext; cachedNext = null; - return retPos; + return currentDelivery; } try @@ -1310,7 +1312,8 @@ final class PageSubscriptionImpl implements PageSubscription position = getStartPosition(); } - return moveNext(); + currentDelivery = moveNext(); + return currentDelivery; } catch (RuntimeException e) { @@ -1473,10 +1476,13 @@ final class PageSubscriptionImpl implements PageSubscription public void remove() { deliveredCount.incrementAndGet(); - PageCursorInfo info = PageSubscriptionImpl.this.getPageInfo(position); - if (info != null) + if (currentDelivery != null) { - info.remove(position); + PageCursorInfo info = PageSubscriptionImpl.this.getPageInfo(currentDelivery.getPosition()); + if (info != null) + { + info.remove(currentDelivery.getPosition()); + } } } diff --git a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/AddressManager.java b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/AddressManager.java index b0b98e8fc5..9ca7dace03 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/AddressManager.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/AddressManager.java @@ -17,6 +17,7 @@ package org.apache.activemq.core.postoffice; import java.util.Map; +import java.util.Set; import org.apache.activemq.api.core.SimpleString; import org.apache.activemq.core.transaction.Transaction; @@ -50,4 +51,6 @@ public interface AddressManager Binding getBinding(SimpleString queueName); Map getBindings(); + + Set getAddresses(); } diff --git a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/PostOffice.java b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/PostOffice.java index de28190b8a..e805d8bb6d 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/PostOffice.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/PostOffice.java @@ -17,6 +17,7 @@ package org.apache.activemq.core.postoffice; import java.util.Map; +import java.util.Set; import org.apache.activemq.api.core.Pair; import org.apache.activemq.api.core.SimpleString; @@ -95,4 +96,6 @@ public interface PostOffice extends ActiveMQComponent boolean isAddressBound(final SimpleString address) throws Exception; + Set getAddresses(); + } diff --git a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/PostOfficeImpl.java b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/PostOfficeImpl.java index 0d0fa036c1..a6b05bfb7f 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/PostOfficeImpl.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/PostOfficeImpl.java @@ -861,6 +861,11 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding return notificationLock; } + public Set getAddresses() + { + return addressManager.getAddresses(); + } + public void sendQueueInfoToQueue(final SimpleString queueName, final SimpleString address) throws Exception { // We send direct to the queue so we can send it to the same queue that is bound to the notifications address - diff --git a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/SimpleAddressManager.java b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/SimpleAddressManager.java index c11bbd537a..259ceb72f7 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/SimpleAddressManager.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/postoffice/impl/SimpleAddressManager.java @@ -16,7 +16,9 @@ */ package org.apache.activemq.core.postoffice.impl; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -151,6 +153,15 @@ public class SimpleAddressManager implements AddressManager mappings.clear(); } + + @Override + public Set getAddresses() + { + Set addresses = new HashSet<>(); + addresses.addAll(mappings.keySet()); + return addresses; + } + protected void removeBindingInternal(final SimpleString address, final SimpleString bindableName) { Bindings bindings = mappings.get(address); diff --git a/activemq-server/src/main/java/org/apache/activemq/core/server/Queue.java b/activemq-server/src/main/java/org/apache/activemq/core/server/Queue.java index 77cd91543d..4759df0f71 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/server/Queue.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/server/Queue.java @@ -176,6 +176,10 @@ public interface Queue extends Bindable boolean checkRedelivery(MessageReference ref, long timeBase, boolean ignoreRedeliveryDelay) throws Exception; + /** + * It will iterate thorugh memory only (not paging) + * @return + */ LinkedListIterator iterator(); LinkedListIterator totalIterator(); @@ -228,7 +232,10 @@ public interface Queue extends Bindable void incrementMesssagesAdded(); - List cancelScheduledMessages(); + /** + * cancels scheduled messages and send them to the head of the queue. + */ + void deliverScheduledMessages(); void postAcknowledge(MessageReference ref); diff --git a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/BackupRecoveryJournalLoader.java b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/BackupRecoveryJournalLoader.java index 32ea1dbf37..39b34df39a 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/BackupRecoveryJournalLoader.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/BackupRecoveryJournalLoader.java @@ -90,7 +90,7 @@ public class BackupRecoveryJournalLoader extends PostOfficeJournalLoader @Override public void postLoad(Journal messageJournal, ResourceManager resourceManager, Map>> duplicateIDMap) throws Exception { - ScaleDownHandler scaleDownHandler = new ScaleDownHandler(pagingManager, postOffice, nodeManager, clusterController); + ScaleDownHandler scaleDownHandler = new ScaleDownHandler(pagingManager, postOffice, nodeManager, clusterController, parentServer.getStorageManager()); locator.setProtocolManagerFactory(ActiveMQServerSideProtocolManagerFactory.getInstance()); try (ClientSessionFactory sessionFactory = locator.createSessionFactory()) diff --git a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/LiveOnlyActivation.java b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/LiveOnlyActivation.java index ff2c8072af..2265b47fed 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/LiveOnlyActivation.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/LiveOnlyActivation.java @@ -187,7 +187,8 @@ public class LiveOnlyActivation extends Activation ScaleDownHandler scaleDownHandler = new ScaleDownHandler(activeMQServer.getPagingManager(), activeMQServer.getPostOffice(), activeMQServer.getNodeManager(), - activeMQServer.getClusterManager().getClusterController()); + activeMQServer.getClusterManager().getClusterController(), + activeMQServer.getStorageManager()); ConcurrentMap duplicateIDCaches = ((PostOfficeImpl) activeMQServer.getPostOffice()).getDuplicateIDCaches(); Map>> duplicateIDMap = new HashMap<>(); for (SimpleString address : duplicateIDCaches.keySet()) diff --git a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/QueueImpl.java b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/QueueImpl.java index 9523205296..e127fe6037 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/QueueImpl.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/QueueImpl.java @@ -1220,9 +1220,18 @@ public class QueueImpl implements Queue } @Override - public List cancelScheduledMessages() + public void deliverScheduledMessages() { - return scheduledDeliveryHandler.cancel(null); + List scheduledMessages = scheduledDeliveryHandler.cancel(null); + if (scheduledMessages != null && scheduledMessages.size() > 0) + { + for (MessageReference ref : scheduledMessages) + { + ref.getMessage().putLongProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME, ref.getScheduledDeliveryTime()); + ref.setScheduledDeliveryTime(0); + } + this.addHead(scheduledMessages); + } } public long getMessagesAdded() @@ -3105,6 +3114,8 @@ public class QueueImpl implements Queue Iterator interIterator = null; LinkedListIterator messagesIterator = null; + Iterator lastIterator = null; + public TotalQueueIterator() { if (pageSubscription != null) @@ -3118,18 +3129,21 @@ public class QueueImpl implements Queue @Override public boolean hasNext() { - if (messagesIterator.hasNext()) + if (messagesIterator != null && messagesIterator.hasNext()) { + lastIterator = messagesIterator; return true; } if (interIterator.hasNext()) { + lastIterator = interIterator; return true; } if (pageIter != null) { if (pageIter.hasNext()) { + lastIterator = pageIter; return true; } } @@ -3140,18 +3154,21 @@ public class QueueImpl implements Queue @Override public MessageReference next() { - if (messagesIterator.hasNext()) + if (messagesIterator != null && messagesIterator.hasNext()) { - return messagesIterator.next(); + MessageReference msg = messagesIterator.next(); + return msg; } if (interIterator.hasNext()) { + lastIterator = interIterator; return interIterator.next(); } if (pageIter != null) { if (pageIter.hasNext()) { + lastIterator = pageIter; return pageIter.next(); } } @@ -3162,6 +3179,10 @@ public class QueueImpl implements Queue @Override public void remove() { + if (lastIterator != null) + { + lastIterator.remove(); + } } @Override @@ -3172,8 +3193,14 @@ public class QueueImpl implements Queue @Override public void close() { - if (pageIter != null) pageIter.close(); - messagesIterator.close(); + if (pageIter != null) + { + pageIter.close(); + } + if (messagesIterator != null) + { + messagesIterator.close(); + } } } diff --git a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/ScaleDownHandler.java b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/ScaleDownHandler.java index 1d351f47b9..f5f87f9724 100644 --- a/activemq-server/src/main/java/org/apache/activemq/core/server/impl/ScaleDownHandler.java +++ b/activemq-server/src/main/java/org/apache/activemq/core/server/impl/ScaleDownHandler.java @@ -20,11 +20,13 @@ import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.Collections; import java.util.Comparator; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.TreeSet; import org.apache.activemq.api.core.Message; import org.apache.activemq.api.core.Pair; @@ -42,7 +44,9 @@ import org.apache.activemq.core.paging.PagingManager; import org.apache.activemq.core.paging.PagingStore; import org.apache.activemq.core.paging.cursor.PageSubscription; import org.apache.activemq.core.paging.cursor.PagedReference; +import org.apache.activemq.core.persistence.StorageManager; import org.apache.activemq.core.postoffice.Binding; +import org.apache.activemq.core.postoffice.Bindings; import org.apache.activemq.core.postoffice.PostOffice; import org.apache.activemq.core.postoffice.impl.LocalQueueBinding; import org.apache.activemq.core.postoffice.impl.PostOfficeImpl; @@ -56,6 +60,7 @@ import org.apache.activemq.core.server.cluster.ClusterController; import org.apache.activemq.core.transaction.ResourceManager; import org.apache.activemq.core.transaction.Transaction; import org.apache.activemq.core.transaction.TransactionOperation; +import org.apache.activemq.core.transaction.impl.TransactionImpl; import org.apache.activemq.utils.LinkedListIterator; public class ScaleDownHandler @@ -64,20 +69,22 @@ public class ScaleDownHandler final PostOffice postOffice; private NodeManager nodeManager; private final ClusterController clusterController; + private final StorageManager storageManager; private String targetNodeId; - public ScaleDownHandler(PagingManager pagingManager, PostOffice postOffice, NodeManager nodeManager, ClusterController clusterController) + public ScaleDownHandler(PagingManager pagingManager, PostOffice postOffice, NodeManager nodeManager, ClusterController clusterController, StorageManager storageManager) { this.pagingManager = pagingManager; this.postOffice = postOffice; this.nodeManager = nodeManager; this.clusterController = clusterController; + this.storageManager = storageManager; } public long scaleDown(ClientSessionFactory sessionFactory, ResourceManager resourceManager, Map>> duplicateIDMap, + List>> duplicateIDMap, SimpleString managementAddress, SimpleString targetNodeId) throws Exception { @@ -91,218 +98,239 @@ public class ScaleDownHandler return num; } - private long scaleDownMessages(ClientSessionFactory sessionFactory, SimpleString nodeId) throws Exception + public long scaleDownMessages(ClientSessionFactory sessionFactory, SimpleString nodeId) throws Exception { long messageCount = 0; targetNodeId = nodeId != null ? nodeId.toString() : getTargetNodeId(sessionFactory); - ClientSession session = sessionFactory.createSession(false, true, true); - Map queueIDs = new HashMap<>(); - ClientProducer producer = session.createProducer(); - - List addresses = new ArrayList<>(); - for (Map.Entry entry : postOffice.getAllBindings().entrySet()) + try (ClientSession session = sessionFactory.createSession(false, true, true)) { - if (entry.getValue() instanceof LocalQueueBinding) - { - SimpleString address = entry.getValue().getAddress(); + ClientProducer producer = session.createProducer(); - // There is a special case involving store-and-forward queues used for clustering. - // If this queue is supposed to forward messages to the server that I'm scaling down to I need to handle these messages differently. - boolean storeAndForward = false; - if (address.toString().startsWith("sf.")) + // perform a loop per address + for (SimpleString address : postOffice.getAddresses()) + { + ActiveMQServerLogger.LOGGER.debug("Scaling down address " + address); + Bindings bindings = postOffice.getBindingsForAddress(address); + + // It will get a list of queues on this address, ordered by the number of messages + Set queues = new TreeSet<>(new OrderQueueByNumberOfReferencesComparator()); + for (Binding binding : bindings.getBindings()) { - // these get special treatment later - storeAndForward = true; + if (binding instanceof LocalQueueBinding) + { + Queue queue = ((LocalQueueBinding) binding).getQueue(); + // as part of scale down we will cancel any scheduled message and pass it to theWhile we scan for the queues we will also cancel any scheduled messages and deliver them right away + queue.deliverScheduledMessages(); + queues.add(queue); + } } - // this means we haven't inspected this address before - if (!addresses.contains(address)) + + if (address.toString().startsWith("sf.")) { - addresses.add(address); + messageCount += scaleDownSNF(address, queues, producer); + } + else + { + messageCount += scaleDownRegularMessages(address, queues, session, producer); + } - PagingStore store = pagingManager.getPageStore(address); + } + } - // compile a list of all the relevant queues and queue iterators for this address - List queues = new ArrayList<>(); - Map> queueIterators = new HashMap<>(); - for (Binding binding : postOffice.getBindingsForAddress(address).getBindings()) + return messageCount; + } + + public long scaleDownRegularMessages(final SimpleString address, final Set queues, final ClientSession clientSession, final ClientProducer producer) throws Exception + { + ActiveMQServerLogger.LOGGER.debug("Scaling down messages on address " + address); + long messageCount = 0; + + final HashMap controls = new HashMap(); + + PagingStore pageStore = pagingManager.getPageStore(address); + + Transaction tx = new TransactionImpl(storageManager); + + pageStore.disableCleanup(); + + try + { + + for (Queue queue : queues) + { + controls.put(queue, new QueuesXRefInnerManager(clientSession, queue, pageStore)); + } + + // compile a list of all the relevant queues and queue iterators for this address + for (Queue loopQueue : queues) + { + ActiveMQServerLogger.LOGGER.debug("Scaling down messages on address " + address + " / performing loop on queue " + loopQueue); + + try (LinkedListIterator messagesIterator = loopQueue.totalIterator()) + { + + while (messagesIterator.hasNext()) { - if (binding instanceof LocalQueueBinding) + MessageReference messageReference = messagesIterator.next(); + Message message = messageReference.getMessage().copy(); + + ActiveMQServerLogger.LOGGER.debug("Reading message " + message + " from queue " + loopQueue); + Set queuesFound = new HashSet<>(); + + for (Map.Entry controlEntry : controls.entrySet()) { - Queue queue = ((LocalQueueBinding) binding).getQueue(); - //remove the scheduled messages and reset on the actual message ready for sending - //we may set the time multiple times on a message but it will always be the same. - //set the ref scheduled time to 0 so it is in the queue ready for resending - List messageReferences = queue.cancelScheduledMessages(); - for (MessageReference ref : messageReferences) + if (controlEntry.getKey() == loopQueue) { - ref.getMessage().putLongProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME, ref.getScheduledDeliveryTime()); - ref.setScheduledDeliveryTime(0); + // no need to lookup on itself, we just add it + queuesFound.add(controlEntry.getValue()); } - queue.addHead(messageReferences); - queues.add(queue); - queueIterators.put(queue.getName(), queue.totalIterator()); + else if (controlEntry.getValue().lookup(messageReference)) + { + ActiveMQServerLogger.LOGGER.debug("Message existed on queue " + controlEntry.getKey().getID() + " removeID=" + controlEntry.getValue().getQueueID()); + queuesFound.add(controlEntry.getValue()); + } + } + + // get the ID for every queue that contains the message + ByteBuffer buffer = ByteBuffer.allocate(queuesFound.size() * 8); + + + for (QueuesXRefInnerManager control : queuesFound) + { + long queueID = control.getQueueID(); + buffer.putLong(queueID); + } + + + message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, buffer.array()); + + if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) + { + if (messageReference.isPaged()) + { + ActiveMQServerLogger.LOGGER.debug("*********************<<<<< Scaling down pdgmessage " + message); + } + else + { + ActiveMQServerLogger.LOGGER.debug("*********************<<<<< Scaling down message " + message); + } + } + + producer.send(address, message); + messageCount++; + + messagesIterator.remove(); + + // We need to perform the ack / removal after sending, otherwise the message could been removed before the send is finished + for (QueuesXRefInnerManager queueFound : queuesFound) + { + ackMessageOnQueue(tx, queueFound.getQueue(), messageReference); + } + + } + } + } + + tx.commit(); + + + for (QueuesXRefInnerManager controlRemoved : controls.values()) + { + controlRemoved.close(); + } + + return messageCount; + } + finally + { + pageStore.enableCleanup(); + pageStore.getCursorProvider().scheduleCleanup(); + } + } + + private long scaleDownSNF(final SimpleString address, final Set queues, final ClientProducer producer) throws Exception + { + long messageCount = 0; + + final String propertyEnd; + + // If this SNF is towards our targetNodeId + boolean queueOnTarget = address.toString().endsWith(targetNodeId); + + if (queueOnTarget) + { + propertyEnd = targetNodeId; + } + else + { + propertyEnd = address.toString().substring(address.toString().lastIndexOf(".")); + } + + Transaction tx = new TransactionImpl(storageManager); + + for (Queue queue : queues) + { + // using auto-closeable + try (LinkedListIterator messagesIterator = queue.totalIterator()) + { + // loop through every message of this queue + while (messagesIterator.hasNext()) + { + MessageReference messageRef = messagesIterator.next(); + Message message = messageRef.getMessage().copy(); + + /* Here we are taking messages out of a store-and-forward queue and sending them to the corresponding + * address on the scale-down target server. However, we have to take the existing _HQ_ROUTE_TOsf.* + * property and put its value into the _HQ_ROUTE_TO property so the message is routed properly. + */ + + byte[] oldRouteToIDs = null; + + List propertiesToRemove = new ArrayList<>(); + message.removeProperty(MessageImpl.HDR_ROUTE_TO_IDS); + for (SimpleString propName : message.getPropertyNames()) + { + if (propName.startsWith(MessageImpl.HDR_ROUTE_TO_IDS)) + { + if (propName.toString().endsWith(propertyEnd)) + { + oldRouteToIDs = message.getBytesProperty(propName); + } + propertiesToRemove.add(propName); } } - // sort into descending order - order is based on the number of references in the queue - Collections.sort(queues, new OrderQueueByNumberOfReferencesComparator()); + // TODO: what if oldRouteToIDs == null ?? - // loop through every queue on this address - List checkedQueues = new ArrayList<>(); - for (Queue bigLoopQueue : queues) + for (SimpleString propertyToRemove : propertiesToRemove) { - checkedQueues.add(bigLoopQueue.getName()); - - LinkedListIterator bigLoopMessageIterator = bigLoopQueue.totalIterator(); - try - { - // loop through every message of this queue - while (bigLoopMessageIterator.hasNext()) - { - MessageReference bigLoopRef = bigLoopMessageIterator.next(); - Message message = bigLoopRef.getMessage().copy(); - - if (storeAndForward) - { - if (address.toString().endsWith(targetNodeId)) - { - /* Here we are taking messages out of a store-and-forward queue and sending them to the corresponding - * address on the scale-down target server. However, we have to take the existing _HQ_ROUTE_TOsf.* - * property and put its value into the _HQ_ROUTE_TO property so the message is routed properly. - */ - - byte[] oldRouteToIDs = null; - - List propertiesToRemove = new ArrayList<>(); - message.removeProperty(MessageImpl.HDR_ROUTE_TO_IDS); - for (SimpleString propName : message.getPropertyNames()) - { - if (propName.startsWith(MessageImpl.HDR_ROUTE_TO_IDS)) - { - if (propName.toString().endsWith(targetNodeId)) - { - oldRouteToIDs = message.getBytesProperty(propName); - } - propertiesToRemove.add(propName); - } - } - - for (SimpleString propertyToRemove : propertiesToRemove) - { - message.removeProperty(propertyToRemove); - } - - message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, oldRouteToIDs); - } - else - { - /* Here we are taking messages out of a store-and-forward queue and sending them to the corresponding - * store-and-forward address on the scale-down target server. In this case we use a special property - * for the queue ID so that the scale-down target server can route it appropriately. - */ - byte[] oldRouteToIDs = null; - - List propertiesToRemove = new ArrayList<>(); - message.removeProperty(MessageImpl.HDR_ROUTE_TO_IDS); - for (SimpleString propName : message.getPropertyNames()) - { - if (propName.startsWith(MessageImpl.HDR_ROUTE_TO_IDS)) - { - if (propName.toString().endsWith(address.toString().substring(address.toString().lastIndexOf(".")))) - { - oldRouteToIDs = message.getBytesProperty(propName); - } - propertiesToRemove.add(propName); - } - } - - for (SimpleString propertyToRemove : propertiesToRemove) - { - message.removeProperty(propertyToRemove); - } - - message.putBytesProperty(MessageImpl.HDR_SCALEDOWN_TO_IDS, oldRouteToIDs); - } - - ActiveMQServerLogger.LOGGER.debug("Scaling down message " + message + " from " + address + " to " + message.getAddress() + " on node " + targetNodeId); - producer.send(message.getAddress(), message); - messageCount++; - bigLoopQueue.deleteReference(message.getMessageID()); - } - else - { - List queuesWithMessage = new ArrayList<>(); - queuesWithMessage.add(bigLoopQueue); - long messageId = message.getMessageID(); - - getQueuesWithMessage(store, queues, queueIterators, checkedQueues, bigLoopQueue, queuesWithMessage, bigLoopRef, messageId); - - // get the ID for every queue that contains the message - ByteBuffer buffer = ByteBuffer.allocate(queuesWithMessage.size() * 8); - StringBuilder logMessage = new StringBuilder(); - logMessage.append("Scaling down message ").append(messageId).append(" to "); - for (Queue queue : queuesWithMessage) - { - long queueID; - String queueName = queue.getName().toString(); - - if (queueIDs.containsKey(queueName)) - { - queueID = queueIDs.get(queueName); - } - else - { - queueID = createQueueIfNecessaryAndGetID(session, queue, address); - queueIDs.put(queueName, queueID); // store it so we don't have to look it up every time - } - - logMessage.append(queueName).append("(").append(queueID).append(")").append(", "); - buffer.putLong(queueID); - } - - logMessage.delete(logMessage.length() - 2, logMessage.length()); // trim off the trailing comma and space - ActiveMQServerLogger.LOGGER.debug(logMessage.append(" on address ").append(address)); - - message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, buffer.array()); - //we need this incase we are sending back to the source server of the message, this basically - //acts like the bridge and ignores dup detection - if (message.containsProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID)) - { - byte[] bytes = new byte[24]; - - ByteBuffer bb = ByteBuffer.wrap(bytes); - bb.put(nodeManager.getUUID().asBytes()); - bb.putLong(messageId); - - message.putBytesProperty(MessageImpl.HDR_BRIDGE_DUPLICATE_ID, bb.array()); - } - - producer.send(address, message); - messageCount++; - - // delete the reference from all queues which contain it - bigLoopQueue.deleteReference(messageId); - for (Queue queue : queuesWithMessage) - { - queue.deleteReference(messageId); - } - } - } - } - finally - { - bigLoopMessageIterator.close(); - queueIterators.get(bigLoopQueue.getName()).close(); - } + message.removeProperty(propertyToRemove); } + + if (queueOnTarget) + { + message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, oldRouteToIDs); + } + else + { + message.putBytesProperty(MessageImpl.HDR_SCALEDOWN_TO_IDS, oldRouteToIDs); + } + + ActiveMQServerLogger.LOGGER.debug("Scaling down message " + message + " from " + address + " to " + message.getAddress() + " on node " + targetNodeId); + producer.send(message.getAddress(), message); + + messageCount++; + + messagesIterator.remove(); + + ackMessageOnQueue(tx, queue, messageRef); } } } - producer.close(); - session.close(); + tx.commit(); return messageCount; } @@ -324,6 +352,8 @@ public class ScaleDownHandler Transaction transaction = resourceManager.getTransaction(xid); session.start(xid, XAResource.TMNOFLAGS); List allOperations = transaction.getAllOperations(); + + // Get the information of the Prepared TXs so it could replay the TXs Map, List>> queuesToSendTo = new HashMap<>(); for (TransactionOperation operation : allOperations) { @@ -387,6 +417,7 @@ public class ScaleDownHandler } } } + ClientProducer producer = session.createProducer(); for (Map.Entry, List>> entry : queuesToSendTo.entrySet()) { @@ -436,78 +467,6 @@ public class ScaleDownHandler } session.close(); } - - /** - * Loop through every *other* queue on this address to see if it also contains this message. - * Skip queues with filters that don't match as matching messages will never be in there. - * Also skip queues that we've already checked in the "big" loop. - */ - private void getQueuesWithMessage(PagingStore store, List queues, Map> queueIterators, List checkedQueues, Queue bigLoopQueue, List queuesWithMessage, MessageReference bigLoopRef, long messageId) throws Exception - { - for (Queue queue : queues) - { - if (!checkedQueues.contains(queue.getName()) && - ((queue.getFilter() == null && - bigLoopQueue.getFilter() == null) || - (queue.getFilter() != null && - queue.getFilter().equals(bigLoopQueue.getFilter())))) - { - // an optimization for paged messages, eliminates the need to (potentially) scan the whole queue - if (bigLoopRef.isPaged()) - { - PageSubscription subscription = store.getCursorProvider().getSubscription(queue.getID()); - if (subscription.contains((PagedReference) bigLoopRef)) - { - queuesWithMessage.add(queue); - } - } - else - { - LinkedListIterator queueIterator = queueIterators.get(queue.getName()); - boolean first = true; - long initialMessageID = 0; - while (queueIterator.hasNext()) - { - Message m = queueIterator.next().getMessage(); - if (first) - { - initialMessageID = m.getMessageID(); - first = false; - } - if (m.getMessageID() == messageId) - { - queuesWithMessage.add(queue); - break; - } - } - - /** - * if we've reached the end then reset the iterator and go through again until we - * get back to the place where we started - */ - if (!queueIterator.hasNext()) - { - queueIterator = queue.totalIterator(); - queueIterators.put(queue.getName(), queueIterator); - while (queueIterator.hasNext()) - { - Message m = queueIterator.next().getMessage(); - if (m.getMessageID() == initialMessageID) - { - break; - } - else if (m.getMessageID() == messageId) - { - queuesWithMessage.add(queue); - break; - } - } - } - } - } - } - } - /** * Get the ID of the queues involved so the message can be routed properly. This is done because we cannot * send directly to a queue, we have to send to an address instead but not all the queues related to the @@ -557,11 +516,159 @@ public class ScaleDownHandler if (queue1 == queue2) return EQUAL; - if (queue1.getMessageCount() == queue2.getMessageCount()) return EQUAL; + if (queue1.getMessageCount() == queue2.getMessageCount()) + { + // if it's the same count we will use the ID as a tie breaker: + + long tieBreak = queue2.getID() - queue1.getID(); + + if (tieBreak > 0) return AFTER; + else if (tieBreak < 0) return BEFORE; + else return EQUAL; // EQUAL here shouldn't really happen... but lets do the check anyways + + } if (queue1.getMessageCount() > queue2.getMessageCount()) return BEFORE; if (queue1.getMessageCount() < queue2.getMessageCount()) return AFTER; return result; } } + + + private void ackMessageOnQueue(Transaction tx, Queue queue, MessageReference messageRef) throws Exception + { + queue.acknowledge(tx, messageRef); + } + + /** + * this class will control iterations while + * looking over for messages relations + */ + private class QueuesXRefInnerManager + { + private final Queue queue; + private LinkedListIterator memoryIterator; + private MessageReference lastRef = null; + private final PagingStore store; + + /** + * ClientSession used for looking up and creating queues + */ + private final ClientSession clientSession; + + private long targetQueueID = -1; + + + QueuesXRefInnerManager(final ClientSession clientSession, final Queue queue, final PagingStore store) + { + this.queue = queue; + this.store = store; + this.clientSession = clientSession; + } + + public Queue getQueue() + { + return queue; + } + + public long getQueueID() throws Exception + { + + if (targetQueueID < 0) + { + targetQueueID = createQueueIfNecessaryAndGetID(clientSession, queue, queue.getAddress()); + } + return targetQueueID; + } + + public void close() + { + if (memoryIterator != null) + { + memoryIterator.close(); + } + } + + public boolean lookup(MessageReference reference) throws Exception + { + + if (reference.isPaged()) + { + PageSubscription subscription = store.getCursorProvider().getSubscription(queue.getID()); + if (subscription.contains((PagedReference) reference)) + { + return true; + } + } + else + { + + if (lastRef != null && lastRef.getMessage().equals(reference.getMessage())) + { + lastRef = null; + memoryIterator.remove(); + return true; + } + + int numberOfScans = 2; + + if (memoryIterator == null) + { + // If we have a brand new iterator, and we can't find something + numberOfScans = 1; + } + + MessageReference initialRef = null; + for (int i = 0; i < numberOfScans; i++) + { + ActiveMQServerLogger.LOGGER.debug("iterating on queue " + queue + " while looking for reference " + reference); + memoryIterator = queue.iterator(); + + while (memoryIterator.hasNext()) + { + lastRef = memoryIterator.next(); + + ActiveMQServerLogger.LOGGER.debug("Iterating on message " + lastRef); + + if (lastRef.getMessage().equals(reference.getMessage())) + { + memoryIterator.remove(); + lastRef = null; + return true; + } + + if (initialRef == null) + { + lastRef = initialRef; + } + else + { + if (initialRef.equals(lastRef)) + { + if (!memoryIterator.hasNext()) + { + // if by coincidence we are at the end of the iterator, we just reset the iterator + lastRef = null; + memoryIterator.close(); + memoryIterator = null; + } + return false; + } + } + } + } + + } + + // if we reached two iterations without finding anything.. we just go away by cleaning everything up + lastRef = null; + memoryIterator.close(); + memoryIterator = null; + + return false; + } + + } + + } diff --git a/activemq-server/src/test/java/org/apache/activemq/core/server/impl/ScheduledDeliveryHandlerTest.java b/activemq-server/src/test/java/org/apache/activemq/core/server/impl/ScheduledDeliveryHandlerTest.java index b41b332a6b..abb6e86d64 100644 --- a/activemq-server/src/test/java/org/apache/activemq/core/server/impl/ScheduledDeliveryHandlerTest.java +++ b/activemq-server/src/test/java/org/apache/activemq/core/server/impl/ScheduledDeliveryHandlerTest.java @@ -1483,9 +1483,9 @@ public class ScheduledDeliveryHandlerTest extends Assert } @Override - public List cancelScheduledMessages() + public void deliverScheduledMessages() { - return null; + } @Override diff --git a/activemq-server/src/test/java/org/apache/activemq/tests/util/ServiceTestBase.java b/activemq-server/src/test/java/org/apache/activemq/tests/util/ServiceTestBase.java index ad621260aa..dd6b9453c0 100644 --- a/activemq-server/src/test/java/org/apache/activemq/tests/util/ServiceTestBase.java +++ b/activemq-server/src/test/java/org/apache/activemq/tests/util/ServiceTestBase.java @@ -972,7 +972,7 @@ public abstract class ServiceTestBase extends UnitTestCase protected HashMap countJournal(Configuration config) throws Exception { final HashMap recordsType = new HashMap(); - SequentialFileFactory messagesFF = new NIOSequentialFileFactory(getJournalDir(), null); + SequentialFileFactory messagesFF = new NIOSequentialFileFactory(config.getJournalDirectory(), null); JournalImpl messagesJournal = new JournalImpl(config.getJournalFileSize(), config.getJournalMinFiles(), diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/HangConsumerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/HangConsumerTest.java index 42c2b59131..b4a79774de 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/HangConsumerTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/HangConsumerTest.java @@ -16,7 +16,6 @@ */ package org.apache.activemq.tests.integration.client; import org.apache.activemq.api.core.ActiveMQException; -import org.apache.activemq.core.server.MessageReference; import org.apache.activemq.core.server.ServerConsumer; import org.junit.Before; import org.junit.After; @@ -24,7 +23,6 @@ import org.junit.Test; import java.lang.management.ManagementFactory; import java.util.LinkedList; -import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; @@ -274,9 +272,8 @@ public class HangConsumerTest extends ServiceTestBase } @Override - public List cancelScheduledMessages() + public void deliverScheduledMessages() { - return null; } } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/InterruptedLargeMessageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/InterruptedLargeMessageTest.java index 25ca4eee63..9db1ff1867 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/InterruptedLargeMessageTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/client/InterruptedLargeMessageTest.java @@ -20,7 +20,6 @@ import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; import java.io.IOException; import java.util.HashMap; -import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; @@ -534,9 +533,8 @@ public class InterruptedLargeMessageTest extends LargeMessageTestBase } @Override - public List cancelScheduledMessages() + public void deliverScheduledMessages() { - return null; } } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/distribution/ClusterTestBase.java b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/distribution/ClusterTestBase.java index f510091ac3..9c4d1185e8 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/distribution/ClusterTestBase.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/distribution/ClusterTestBase.java @@ -62,6 +62,7 @@ import org.apache.activemq.core.postoffice.Bindings; import org.apache.activemq.core.postoffice.PostOffice; import org.apache.activemq.core.postoffice.QueueBinding; import org.apache.activemq.core.postoffice.impl.LocalQueueBinding; +import org.apache.activemq.core.protocol.core.impl.CoreProtocolManagerFactory; import org.apache.activemq.core.remoting.impl.netty.TransportConstants; import org.apache.activemq.core.server.ActiveMQServer; import org.apache.activemq.core.server.ActiveMQServers; @@ -1681,7 +1682,8 @@ public abstract class ClusterTestBase extends ServiceTestBase .setThreadPoolMaxSize(10) .clearAcceptorConfigurations() .addAcceptorConfiguration(createTransportConfiguration(netty, true, generateParams(node, netty))) - .setHAPolicyConfiguration(haPolicyConfiguration); + .setHAPolicyConfiguration(haPolicyConfiguration) + .setResolveProtocols(false); ActiveMQServer server; @@ -1708,6 +1710,8 @@ public abstract class ClusterTestBase extends ServiceTestBase } } + server.addProtocolManagerFactory(new CoreProtocolManagerFactory()); + server.setIdentity(this.getClass().getSimpleName() + "/Live(" + node + ")"); servers[node] = server; } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownDirectTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownDirectTest.java new file mode 100644 index 0000000000..0ae56db919 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownDirectTest.java @@ -0,0 +1,319 @@ +/** + * 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.tests.integration.server; + +import java.util.Arrays; +import java.util.Collection; + +import org.apache.activemq.api.core.Message; +import org.apache.activemq.api.core.SimpleString; +import org.apache.activemq.api.core.client.ActiveMQClient; +import org.apache.activemq.api.core.client.ClientConsumer; +import org.apache.activemq.api.core.client.ClientMessage; +import org.apache.activemq.api.core.client.ClientProducer; +import org.apache.activemq.api.core.client.ClientSession; +import org.apache.activemq.api.core.client.ClientSessionFactory; +import org.apache.activemq.core.postoffice.impl.LocalQueueBinding; +import org.apache.activemq.core.server.impl.ScaleDownHandler; +import org.apache.activemq.core.settings.impl.AddressSettings; +import org.apache.activemq.tests.integration.cluster.distribution.ClusterTestBase; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +/** + * On this test we will run ScaleDown directly as an unit-test in several cases, + * simulating what would happen during a real scale down. + * + * @author clebertsuconic + */ +@RunWith(value = Parameterized.class) +public class ScaleDownDirectTest extends ClusterTestBase +{ + + + @Parameterized.Parameters(name = "isNetty={0}") + public static Collection getParameters() + { + return Arrays.asList(new Object[][]{ + {false}, {true} + }); + } + + + private final boolean isNetty; + + public ScaleDownDirectTest(boolean isNetty) + { + this.isNetty = isNetty; + } + + @Override + @Before + public void setUp() throws Exception + { + super.setUp(); + setupLiveServer(0, isFileStorage(), isNetty, true); + setupLiveServer(1, isFileStorage(), isNetty, true); + startServers(0, 1); + setupSessionFactory(0, isNetty); + setupSessionFactory(1, isNetty); + + } + + @Override + @After + public void tearDown() throws Exception + { + super.tearDown(); + } + + @Test + public void testSendMixedSmallMessages() throws Exception + { + internalTest(100, 100); + } + + @Test + public void testSendMixedLargelMessages() throws Exception + { + internalTest(2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, 100); + } + + protected void internalTest(int bufferSize, int numberOfMessages) throws Exception + { + ClientSessionFactory sf = sfs[0]; + + ClientSession session = sf.createSession(true, true); + + session.createQueue("ad1", "queue1", true); + + ClientProducer producer = session.createProducer("ad1"); + + byte[] buffer = new byte[bufferSize]; + for (int i = 0; i < bufferSize; i++) + { + buffer[i] = getSamplebyte(i); + } + + for (int i = 0; i < numberOfMessages; i++) + { + ClientMessage message = session.createMessage(true); + message.putIntProperty("i", i); + message.getBodyBuffer().writeBytes(buffer); + producer.send(message); + } + + session.createQueue("ad1", "queue2", true); + + for (int i = numberOfMessages; i < (numberOfMessages * 2); i++) + { + ClientMessage message = session.createMessage(true); + message.putIntProperty("i", i); + message.getBodyBuffer().writeBytes(buffer); + producer.send(message); + } + + assertEquals(numberOfMessages * 2, performScaledown()); + + sfs[0].close(); + + session.close(); + + stopServers(0); + + session = sfs[1].createSession(true, true); + + ClientConsumer consumer1 = session.createConsumer("queue1"); + session.start(); + + + for (int i = 0; i < numberOfMessages * 2; i++) + { + ClientMessage message = consumer1.receive(5000); + assertNotNull(message); + assertEquals(i, message.getIntProperty("i").intValue()); +// message.acknowledge(); + + checkBody(message, bufferSize); + + } + + ClientMessage messageCheckNull = consumer1.receiveImmediate(); + + assertNull(messageCheckNull); + + ClientConsumer consumer2 = session.createConsumer("queue2"); + for (int i = numberOfMessages; i < numberOfMessages * 2; i++) + { + ClientMessage message = consumer2.receive(5000); + assertNotNull(message); + assertEquals(i, message.getIntProperty("i").intValue()); +// message.acknowledge(); + checkBody(message, bufferSize); + } + + messageCheckNull = consumer2.receiveImmediate(); + + System.out.println("Received " + messageCheckNull); + + assertNull(messageCheckNull); + } + + + + @Test + public void testPaging() throws Exception + { + final int CHUNK_SIZE = 50; + int messageCount = 0; + final String addressName = "testAddress"; + final String queueName = "testQueue"; + + createQueue(0, addressName, queueName, null, true); + createQueue(1, addressName, queueName, null, true); + + ClientSessionFactory sf = sfs[0]; + ClientSession session = addClientSession(sf.createSession(false, false)); + ClientProducer producer = addClientProducer(session.createProducer(addressName)); + + AddressSettings defaultSetting = new AddressSettings(); + defaultSetting.setPageSizeBytes(10 * 1024); + defaultSetting.setMaxSizeBytes(20 * 1024); + servers[0].getAddressSettingsRepository().addMatch("#", defaultSetting); + + while (!servers[0].getPagingManager().getPageStore(new SimpleString(addressName)).isPaging()) + { + for (int i = 0; i < CHUNK_SIZE; i++) + { + Message message = session.createMessage(true); + message.getBodyBuffer().writeBytes(new byte[1024]); + // The only purpose of this count here is for eventually debug messages on print-data / print-pages +// message.putIntProperty("count", messageCount); + producer.send(message); + messageCount++; + } + session.commit(); + } + + assertEquals(messageCount, performScaledown()); + + servers[0].stop(); + + addConsumer(0, 1, queueName, null); + for (int i = 0; i < messageCount; i++) + { + ClientMessage message = consumers[0].getConsumer().receive(500); + Assert.assertNotNull(message); +// Assert.assertEquals(i, message.getIntProperty("count").intValue()); + } + + Assert.assertNull(consumers[0].getConsumer().receiveImmediate()); + removeConsumer(0); + } + + + + @Test + public void testBasicScaleDown() throws Exception + { + final int TEST_SIZE = 2; + final String addressName = "testAddress"; + final String queueName1 = "testQueue1"; + final String queueName2 = "testQueue2"; + + // create 2 queues on each node mapped to the same address + createQueue(0, addressName, queueName1, null, true); + createQueue(0, addressName, queueName2, null, true); + createQueue(1, addressName, queueName1, null, true); + createQueue(1, addressName, queueName2, null, true); + + // send messages to node 0 + send(0, addressName, TEST_SIZE, true, null); + + // consume a message from queue 2 + addConsumer(1, 0, queueName2, null, false); + ClientMessage clientMessage = consumers[1].getConsumer().receive(250); + Assert.assertNotNull(clientMessage); + clientMessage.acknowledge(); + consumers[1].getSession().commit(); + removeConsumer(1); + + // at this point on node 0 there should be 2 messages in testQueue1 and 1 message in testQueue2 + Assert.assertEquals(TEST_SIZE, getMessageCount(((LocalQueueBinding) servers[0].getPostOffice().getBinding(new SimpleString(queueName1))).getQueue())); + Assert.assertEquals(TEST_SIZE - 1, getMessageCount(((LocalQueueBinding) servers[0].getPostOffice().getBinding(new SimpleString(queueName2))).getQueue())); + + assertEquals(TEST_SIZE, performScaledown()); + // trigger scaleDown from node 0 to node 1 + servers[0].stop(); + + // get the 2 messages from queue 1 + addConsumer(0, 1, queueName1, null); + clientMessage = consumers[0].getConsumer().receive(250); + Assert.assertNotNull(clientMessage); + clientMessage.acknowledge(); + clientMessage = consumers[0].getConsumer().receive(250); + Assert.assertNotNull(clientMessage); + clientMessage.acknowledge(); + + // ensure there are no more messages on queue 1 + clientMessage = consumers[0].getConsumer().receive(250); + Assert.assertNull(clientMessage); + removeConsumer(0); + + // get the 1 message from queue 2 + addConsumer(0, 1, queueName2, null); + clientMessage = consumers[0].getConsumer().receive(250); + Assert.assertNotNull(clientMessage); + clientMessage.acknowledge(); + + // ensure there are no more messages on queue 1 + clientMessage = consumers[0].getConsumer().receive(250); + Assert.assertNull(clientMessage); + removeConsumer(0); + } + + private void checkBody(ClientMessage message, int bufferSize) + { + assertEquals(bufferSize, message.getBodySize()); + byte[] body = new byte[message.getBodySize()]; + message.getBodyBuffer().readBytes(body); + for (int bpos = 0; bpos < bufferSize; bpos++) + { + if (getSamplebyte(bpos) != body[bpos]) + { + fail("body comparison failure at " + message); + } + } + } + + private long performScaledown() throws Exception + { + ScaleDownHandler handler = new ScaleDownHandler(servers[0].getPagingManager(), servers[0].getPostOffice(), + servers[0].getNodeManager(), + servers[0].getClusterManager().getClusterController(), + servers[0].getStorageManager()); + + return handler.scaleDownMessages(sfs[1], servers[1].getNodeID()); + } + + +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownTest.java index 72666c7d76..9119927812 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/server/ScaleDownTest.java @@ -30,8 +30,6 @@ import org.apache.activemq.api.core.client.ClientSessionFactory; import org.apache.activemq.api.core.client.ActiveMQClient; import org.apache.activemq.core.config.ScaleDownConfiguration; import org.apache.activemq.core.config.ha.LiveOnlyPolicyConfiguration; -import org.apache.activemq.core.persistence.impl.journal.JournalStorageManager; -import org.apache.activemq.core.persistence.impl.journal.LargeServerMessageImpl; import org.apache.activemq.core.postoffice.Binding; import org.apache.activemq.core.postoffice.impl.LocalQueueBinding; import org.apache.activemq.core.settings.impl.AddressSettings; @@ -119,13 +117,13 @@ public class ScaleDownTest extends ClusterTestBase final String queueName2 = "testQueue2"; // create 2 queues on each node mapped to the same address - createQueue(0, addressName, queueName1, null, false); - createQueue(0, addressName, queueName2, null, false); - createQueue(1, addressName, queueName1, null, false); - createQueue(1, addressName, queueName2, null, false); + createQueue(0, addressName, queueName1, null, true); + createQueue(0, addressName, queueName2, null, true); + createQueue(1, addressName, queueName1, null, true); + createQueue(1, addressName, queueName2, null, true); // send messages to node 0 - send(0, addressName, TEST_SIZE, false, null); + send(0, addressName, TEST_SIZE, true, null); // consume a message from queue 2 addConsumer(1, 0, queueName2, null, false); @@ -375,32 +373,28 @@ public class ScaleDownTest extends ClusterTestBase final String addressName = "testAddress"; final String queueName = "testQueue"; - createQueue(0, addressName, queueName, null, false); - createQueue(1, addressName, queueName, null, false); + createQueue(0, addressName, queueName, null, true); + createQueue(1, addressName, queueName, null, true); ClientSessionFactory sf = sfs[0]; ClientSession session = addClientSession(sf.createSession(false, false)); ClientProducer producer = addClientProducer(session.createProducer(addressName)); - LargeServerMessageImpl fileMessage = new LargeServerMessageImpl((JournalStorageManager) servers[0].getStorageManager()); - fileMessage.setMessageID(1005); - fileMessage.setDurable(true); - - for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) + byte[] buffer = new byte[2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE]; + for (int i = 0; i < buffer.length; i++) { - fileMessage.addBytes(new byte[]{UnitTestCase.getSamplebyte(i)}); + buffer[i] = getSamplebyte(i); } - fileMessage.putLongProperty(Message.HDR_LARGE_BODY_SIZE, 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE); + for (int nmsg = 0; nmsg < 10; nmsg++) + { + ClientMessage message = session.createMessage(true); + message.getBodyBuffer().writeBytes(buffer); + producer.send(message); + session.commit(); + } - fileMessage.releaseResources(); - - producer.send(fileMessage); - - fileMessage.deleteFile(); - - session.commit(); servers[0].stop(); @@ -409,19 +403,23 @@ public class ScaleDownTest extends ClusterTestBase ClientConsumer consumer = addClientConsumer(session.createConsumer(queueName)); session.start(); - ClientMessage msg = consumer.receive(250); - - Assert.assertNotNull(msg); - - Assert.assertEquals(2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, msg.getBodySize()); - - for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) + for (int nmsg = 0; nmsg < 10; nmsg++) { - Assert.assertEquals(UnitTestCase.getSamplebyte(i), msg.getBodyBuffer().readByte()); - } + ClientMessage msg = consumer.receive(250); - msg.acknowledge(); - session.commit(); + Assert.assertNotNull(msg); + + Assert.assertEquals(2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, msg.getBodySize()); + + for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) + { + byte byteRead = msg.getBodyBuffer().readByte(); + Assert.assertEquals(msg + " Is different", UnitTestCase.getSamplebyte(i), byteRead); + } + + msg.acknowledge(); + session.commit(); + } } @Test diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/postoffice/impl/FakeQueue.java b/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/postoffice/impl/FakeQueue.java index e4c4efd4d7..b88ee6ea6e 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/postoffice/impl/FakeQueue.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/postoffice/impl/FakeQueue.java @@ -362,9 +362,9 @@ public class FakeQueue implements Queue } @Override - public List cancelScheduledMessages() + public void deliverScheduledMessages() { - return null; + } @Override diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/server/impl/fakes/FakePostOffice.java b/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/server/impl/fakes/FakePostOffice.java index cf431d4d57..70cdb7f04f 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/server/impl/fakes/FakePostOffice.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/tests/unit/core/server/impl/fakes/FakePostOffice.java @@ -17,6 +17,7 @@ package org.apache.activemq.tests.unit.core.server.impl.fakes; import java.util.Map; +import java.util.Set; import org.apache.activemq.api.core.Pair; import org.apache.activemq.api.core.SimpleString; @@ -43,6 +44,12 @@ public class FakePostOffice implements PostOffice return false; } + @Override + public Set getAddresses() + { + return null; + } + @Override public void start() throws Exception {