This is an automated email from the ASF dual-hosted git repository. mattrpav pushed a commit to branch activemq-5.19.x in repository https://gitbox.apache.org/repos/asf/activemq.git
commit ee08feec8c374e522f9292fb57e1dec704be55d6 Author: Matt Pavlovich <[email protected]> AuthorDate: Sat Jul 18 09:02:37 2026 -0500 [#2230] KahaDB optionally use a isolatedIterator with isolatedCursor (cherry picked from commit ed7d31b7345b8e3f37b739873d88b8414c05759d) --- .../activemq/store/MessageRecoveryContext.java | 30 +- .../activemq/store/MessageRecoveryContextTest.java | 10 +- .../apache/activemq/store/kahadb/KahaDBStore.java | 43 ++- .../activemq/store/kahadb/MessageDatabase.java | 200 ++++++++----- .../KahaDBRecoverMessagesIsolatedCursorTest.java | 329 +++++++++++++++++++++ 5 files changed, 512 insertions(+), 100 deletions(-) diff --git a/activemq-broker/src/main/java/org/apache/activemq/store/MessageRecoveryContext.java b/activemq-broker/src/main/java/org/apache/activemq/store/MessageRecoveryContext.java index da2b198ee2..99e841bbbc 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/store/MessageRecoveryContext.java +++ b/activemq-broker/src/main/java/org/apache/activemq/store/MessageRecoveryContext.java @@ -23,10 +23,10 @@ import org.apache.activemq.command.MessageId; public class MessageRecoveryContext implements MessageRecoveryListener { public static final int DEFAULT_MAX_MESSAGE_COUNT_RETURNED = 100; - public static final boolean DEFAULT_USE_DEDICATED_CURSOR = true; + public static final boolean DEFAULT_USE_ISOLATED_CURSOR = true; // Config - private final boolean useDedicatedCursor; + private final boolean useIsolatedCursor; private final int maxMessageCountReturned; private final Long offset; private final String startMessageId; @@ -41,7 +41,7 @@ public class MessageRecoveryContext implements MessageRecoveryListener { MessageRecoveryContext(final MessageRecoveryListener messageRecoveryListener, final String startMessageId, final String endMessageId, final Long offset, final Integer maxMessageCountReturned, - final Boolean useDedicatedCursor) { + final Boolean useIsolatedCursor) { if(maxMessageCountReturned != null && maxMessageCountReturned < 0) { throw new IllegalArgumentException("maxMessageCountReturned must be a positive integer value"); } @@ -61,11 +61,16 @@ public class MessageRecoveryContext implements MessageRecoveryListener { this.messageRecoveryListener = messageRecoveryListener; this.offset = offset; this.startMessageId = startMessageId; - this.useDedicatedCursor = (useDedicatedCursor != null ? useDedicatedCursor : DEFAULT_USE_DEDICATED_CURSOR); + this.useIsolatedCursor = (useIsolatedCursor != null ? useIsolatedCursor : DEFAULT_USE_ISOLATED_CURSOR); } + public boolean isUseIsolatedCursor() { + return this.useIsolatedCursor; + } + + @Deprecated(forRemoval = true) public boolean isUseDedicatedCursor() { - return this.useDedicatedCursor; + return isUseIsolatedCursor(); } public int getMaxMessageCountReturned() { @@ -133,7 +138,7 @@ public class MessageRecoveryContext implements MessageRecoveryListener { @Override public String toString() { - return "MessageRecoveryContext [useDedicatedCursor=" + useDedicatedCursor + ", maxMessageCountReturned=" + return "MessageRecoveryContext [useIsolatedCursor=" + useIsolatedCursor + ", maxMessageCountReturned=" + maxMessageCountReturned + ", offset=" + offset + ", startMessageId=" + startMessageId + ", endMessageId=" + endMessageId + ", messageRecoveryListener=" + messageRecoveryListener + ", endSequenceId=" + endSequenceId + ", recoveredCount=" + recoveredCount + "]"; @@ -141,18 +146,23 @@ public class MessageRecoveryContext implements MessageRecoveryListener { public static class Builder { - private Boolean useDedicatedCursor; + private Boolean useIsolatedCursor; private Integer maxMessageCountReturned; private Long offset; private String startMessageId; private String endMessageId; private MessageRecoveryListener messageRecoveryListener; - public Builder useDedicatedCursor(final boolean useDedicatedCursor) { - this.useDedicatedCursor = useDedicatedCursor; + public Builder useIsolatedCursor(final boolean useIsolatedCursor) { + this.useIsolatedCursor = useIsolatedCursor; return this; } + @Deprecated(forRemoval = true) + public Builder useDedicatedCursor(final boolean useDedicatedCursor) { + return useIsolatedCursor(useDedicatedCursor); + } + public Builder maxMessageCountReturned(final int maxMessageCountReturned) { this.maxMessageCountReturned = maxMessageCountReturned; return this; @@ -179,7 +189,7 @@ public class MessageRecoveryContext implements MessageRecoveryListener { } public MessageRecoveryContext build() { - return new MessageRecoveryContext(messageRecoveryListener, startMessageId, endMessageId, offset, maxMessageCountReturned, useDedicatedCursor); + return new MessageRecoveryContext(messageRecoveryListener, startMessageId, endMessageId, offset, maxMessageCountReturned, useIsolatedCursor); } } } diff --git a/activemq-broker/src/test/java/org/apache/activemq/store/MessageRecoveryContextTest.java b/activemq-broker/src/test/java/org/apache/activemq/store/MessageRecoveryContextTest.java index f77c94b5cd..714f441da6 100644 --- a/activemq-broker/src/test/java/org/apache/activemq/store/MessageRecoveryContextTest.java +++ b/activemq-broker/src/test/java/org/apache/activemq/store/MessageRecoveryContextTest.java @@ -39,17 +39,17 @@ public class MessageRecoveryContextTest { assertNotNull(messageRecoveryContext.getMessageRecoveryListener()); assertEquals(Long.valueOf(10_000l), Long.valueOf(messageRecoveryContext.getOffset())); assertNull(messageRecoveryContext.getStartMessageId()); - assertTrue(messageRecoveryContext.isUseDedicatedCursor()); + assertTrue(messageRecoveryContext.isUseIsolatedCursor()); } @Test - public void testConfigOffsetNoDedicatedCursor() { + public void testConfigOffsetNoIsolatedCursor() { MessageRecoveryContext messageRecoveryContext = new MessageRecoveryContext.Builder() .maxMessageCountReturned(999) .messageRecoveryListener(new TestMessageRecoveryListener()) .offset(10_000) - .useDedicatedCursor(false) + .useIsolatedCursor(false) .build(); assertNotNull(messageRecoveryContext); @@ -58,7 +58,7 @@ public class MessageRecoveryContextTest { assertNotNull(messageRecoveryContext.getMessageRecoveryListener()); assertEquals(Long.valueOf(10_000l), Long.valueOf(messageRecoveryContext.getOffset())); assertNull(messageRecoveryContext.getStartMessageId()); - assertFalse(messageRecoveryContext.isUseDedicatedCursor()); + assertFalse(messageRecoveryContext.isUseIsolatedCursor()); } @Test @@ -77,7 +77,7 @@ public class MessageRecoveryContextTest { assertNotNull(messageRecoveryContext.getMessageRecoveryListener()); assertNull(messageRecoveryContext.getOffset()); assertEquals("ID-start-12", messageRecoveryContext.getStartMessageId()); - assertTrue(messageRecoveryContext.isUseDedicatedCursor()); + assertTrue(messageRecoveryContext.isUseIsolatedCursor()); } @Test(expected = IllegalArgumentException.class) diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java index 5998dcfaab..face373bfa 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java @@ -656,9 +656,8 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, pageFile.tx().execute(tx -> { StoredDestination sd = getStoredDestination(dest, tx); recoverRolledBackAcks(destination.getPhysicalName(), sd, tx, Integer.MAX_VALUE, listener); - sd.orderIndex.resetCursorPosition(); for (Iterator<Entry<Long, MessageKeys>> iterator = - sd.orderIndex.iterator(tx, new MessageOrderCursor()); listener.hasSpace() && + sd.orderIndex.iteratorIsolated(tx); listener.hasSpace() && iterator.hasNext(); ) { Entry<Long, MessageKeys> entry = iterator.next(); Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName()); @@ -703,10 +702,28 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, } } + /** + * Recover messages non-destructively by offset or messageId range. + * + * <p>With {@code useIsolatedCursor=true} (the default) the iteration is + * fully isolated from the destination: it does not move the destination + * cursor, does not record iteration bookmarks, and does not consume + * rolled-back transactional acks pending redelivery through the live + * cursor. With {@code useIsolatedCursor=false} the iteration behaves + * like a live batch: it starts at the destination cursor (the start + * offset/messageId is ignored), replays rolled-back acks, and advances + * the destination cursor when done. + * + * <p>Range semantics: the {@code startMessageId} (or {@code offset}) and + * {@code endMessageId} are both inclusive. A {@code startMessageId} that + * is not found in the index falls back to the head of the store; an + * {@code endMessageId} that is not found falls back to + * {@code start + maxMessageCountReturned}. + */ @Override public void recoverMessages(final MessageRecoveryContext messageRecoveryContext) throws Exception { - if(messageRecoveryContext == null || + if(messageRecoveryContext == null || (messageRecoveryContext.getStartMessageId() != null && messageRecoveryContext.getOffset() != null)) { LOG.warn("Invalid messageRecoveryContext:{}", messageRecoveryContext); @@ -759,10 +776,19 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, } Entry<Long, MessageKeys> entry; - recoverRolledBackAcks(destination.getPhysicalName(), sd, tx, messageRecoveryContext.getMaxMessageCountReturned(), messageRecoveryContext); + // [AMQ-9773] Rolled-back acks are pending redelivery through the live cursor. + // Only replay them in shared-cursor mode (live-cursor semantics); an isolated-cursor + // visit must not consume them or the live cursor never redelivers those messages. + if(!messageRecoveryContext.isUseIsolatedCursor()) { + recoverRolledBackAcks(destination.getPhysicalName(), sd, tx, messageRecoveryContext.getMaxMessageCountReturned(), messageRecoveryContext); + } Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName()); - Iterator<Entry<Long, MessageKeys>> iterator = (messageRecoveryContext.isUseDedicatedCursor() ? sd.orderIndex.iterator(tx, - new MessageOrderCursor(startSequenceOffset)) : sd.orderIndex.iterator(tx)); + // [AMQ-9773] An isolated cursor uses IsolatedMessageOrderIterator so no lastXxxKey + // bookmarks are recorded on the shared order index — a later zero-entry live batch + // would otherwise commit them via stoppedIterating() and corrupt the destination cursor. + Iterator<Entry<Long, MessageKeys>> iterator = (messageRecoveryContext.isUseIsolatedCursor() ? + sd.orderIndex.iteratorIsolated(tx, new MessageOrderCursor(startSequenceOffset)) : + sd.orderIndex.iterator(tx)); while (iterator.hasNext()) { entry = iterator.next(); @@ -781,7 +807,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, } // [AMQ-9773] The sd.orderIndex uses the destination's cursor - if(!messageRecoveryContext.isUseDedicatedCursor()) { + if(!messageRecoveryContext.isUseIsolatedCursor()) { sd.orderIndex.stoppedIterating(); } }); @@ -1280,7 +1306,6 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, try { return pageFile.tx().execute(tx -> { StoredDestination sd = getStoredDestination(dest, tx); - sd.orderIndex.resetCursorPosition(); int count = 0; final Map<SubscriptionKey, List<Message>> expired = new HashMap<>(); final Map<String, SubscriptionKey> subKeys = new HashMap<>(); @@ -1297,7 +1322,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, // hit the max browse limit, or if the listener returns false for hasSpace() final Set<Long> uniqueExpired = new HashSet<>(); for (Iterator<Entry<Long, MessageKeys>> iterator = - sd.orderIndex.iterator(tx, new MessageOrderCursor()); count < maxBrowse && iterator.hasNext() && listener.hasSpace(); ) { + sd.orderIndex.iteratorIsolated(tx); count < maxBrowse && iterator.hasNext() && listener.hasSpace(); ) { count++; Entry<Long, MessageKeys> entry = iterator.next(); Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName()); diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java index 058e1fe321..425337303f 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java @@ -3843,6 +3843,12 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } + /** + * Iterate and record the last visited sequence keys + * (lastDefaultKey/lastHighKey/lastLowKey) so that a subsequent + * {@link #stoppedIterating()} advances the destination cursor past the + * visited messages. + */ Iterator<Entry<Long, MessageKeys>> iterator(Transaction tx) throws IOException{ return new MessageOrderIterator(tx,cursor,this); } @@ -3851,6 +3857,14 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe return new MessageOrderIterator(tx,m,this); } + Iterator<Entry<Long, MessageKeys>> iteratorIsolated(Transaction tx) throws IOException{ + return iteratorIsolated(tx, new MessageOrderCursor()); + } + + Iterator<Entry<Long, MessageKeys>> iteratorIsolated(Transaction tx, MessageOrderCursor m) throws IOException{ + return new IsolatedMessageOrderIterator(tx, m, this); + } + public byte lastGetPriority() { return lastGetPriority; } @@ -3883,100 +3897,134 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } - class MessageOrderIterator implements Iterator<Entry<Long, MessageKeys>>{ - Iterator<Entry<Long, MessageKeys>>currentIterator; - final Iterator<Entry<Long, MessageKeys>>highIterator; - final Iterator<Entry<Long, MessageKeys>>defaultIterator; - final Iterator<Entry<Long, MessageKeys>>lowIterator; + } - MessageOrderIterator(Transaction tx, MessageOrderCursor m, MessageOrderIndex messageOrderIndex) throws IOException { - Long pendingAddLimiter = messageOrderIndex.minPendingAdd(); - this.defaultIterator = defaultPriorityIndex.iterator(tx, m.defaultCursorPosition, pendingAddLimiter); - if (highPriorityIndex != null) { - this.highIterator = highPriorityIndex.iterator(tx, m.highPriorityCursorPosition, pendingAddLimiter); - } else { - this.highIterator = null; - } - if (lowPriorityIndex != null) { - this.lowIterator = lowPriorityIndex.iterator(tx, m.lowPriorityCursorPosition, pendingAddLimiter); - } else { - this.lowIterator = null; - } + /** + * Read-only iterator over a {@link MessageOrderIndex}'s three priority + * tiers (high, default, low). Does not record lastDefaultKey / + * lastHighKey / lastLowKey bookmarks, so a subsequent + * {@link MessageOrderIndex#stoppedIterating()} is a no-op with respect + * to cursor advancement. + * + * <p>Use this variant for browses, statistics scans and any other + * iteration that must not affect the destination cursor. + */ + static class IsolatedMessageOrderIterator implements Iterator<Entry<Long, MessageKeys>> { + Iterator<Entry<Long, MessageKeys>> currentIterator; + final Iterator<Entry<Long, MessageKeys>> highIterator; + final Iterator<Entry<Long, MessageKeys>> defaultIterator; + final Iterator<Entry<Long, MessageKeys>> lowIterator; + + IsolatedMessageOrderIterator(Transaction tx, MessageOrderCursor m, + MessageOrderIndex orderIndex) throws IOException { + Long pendingAddLimiter = orderIndex.minPendingAdd(); + this.defaultIterator = orderIndex.defaultPriorityIndex.iterator(tx, m.defaultCursorPosition, pendingAddLimiter); + if (orderIndex.highPriorityIndex != null) { + this.highIterator = orderIndex.highPriorityIndex.iterator(tx, m.highPriorityCursorPosition, pendingAddLimiter); + } else { + this.highIterator = null; } + if (orderIndex.lowPriorityIndex != null) { + this.lowIterator = orderIndex.lowPriorityIndex.iterator(tx, m.lowPriorityCursorPosition, pendingAddLimiter); + } else { + this.lowIterator = null; + } + } - @Override - public boolean hasNext() { - if (currentIterator == null) { - if (highIterator != null) { - if (highIterator.hasNext()) { - currentIterator = highIterator; - return currentIterator.hasNext(); - } - if (defaultIterator.hasNext()) { - currentIterator = defaultIterator; - return currentIterator.hasNext(); - } - if (lowIterator.hasNext()) { - currentIterator = lowIterator; - return currentIterator.hasNext(); - } - return false; - } else { + @Override + public boolean hasNext() { + if (currentIterator == null) { + if (highIterator != null) { + if (highIterator.hasNext()) { + currentIterator = highIterator; + return currentIterator.hasNext(); + } + if (defaultIterator.hasNext()) { currentIterator = defaultIterator; return currentIterator.hasNext(); } + if (lowIterator.hasNext()) { + currentIterator = lowIterator; + return currentIterator.hasNext(); + } + return false; + } else { + currentIterator = defaultIterator; + return currentIterator.hasNext(); } - if (highIterator != null) { - if (currentIterator.hasNext()) { - return true; + } + if (highIterator != null) { + if (currentIterator.hasNext()) { + return true; + } + if (currentIterator == highIterator) { + if (defaultIterator.hasNext()) { + currentIterator = defaultIterator; + return currentIterator.hasNext(); } - if (currentIterator == highIterator) { - if (defaultIterator.hasNext()) { - currentIterator = defaultIterator; - return currentIterator.hasNext(); - } - if (lowIterator.hasNext()) { - currentIterator = lowIterator; - return currentIterator.hasNext(); - } - return false; + if (lowIterator.hasNext()) { + currentIterator = lowIterator; + return currentIterator.hasNext(); } + return false; + } - if (currentIterator == defaultIterator) { - if (lowIterator.hasNext()) { - currentIterator = lowIterator; - return currentIterator.hasNext(); - } - return false; + if (currentIterator == defaultIterator) { + if (lowIterator.hasNext()) { + currentIterator = lowIterator; + return currentIterator.hasNext(); } + return false; } - return currentIterator.hasNext(); } + return currentIterator.hasNext(); + } - @Override - public Entry<Long, MessageKeys> next() { - Entry<Long, MessageKeys> result = currentIterator.next(); - if (result != null) { - Long key = result.getKey(); - if (highIterator != null) { - if (currentIterator == defaultIterator) { - lastDefaultKey = key; - } else if (currentIterator == highIterator) { - lastHighKey = key; - } else { - lastLowKey = key; - } + @Override + public Entry<Long, MessageKeys> next() { + return currentIterator.next(); + } + + @Override + public void remove() { + throw new UnsupportedOperationException(); + } + } + + /** + * Iterator that extends {@link IsolatedMessageOrderIterator} to record + * the last visited sequence key per priority tier into the parent + * {@link MessageOrderIndex}. A subsequent + * {@link MessageOrderIndex#stoppedIterating()} commits those bookmarks + * into the destination cursor, advancing it past the visited messages. + */ + static class MessageOrderIterator extends IsolatedMessageOrderIterator { + private final MessageOrderIndex orderIndex; + + MessageOrderIterator(Transaction tx, MessageOrderCursor m, + MessageOrderIndex orderIndex) throws IOException { + super(tx, m, orderIndex); + this.orderIndex = orderIndex; + } + + @Override + public Entry<Long, MessageKeys> next() { + Entry<Long, MessageKeys> result = super.next(); + if (result != null) { + Long key = result.getKey(); + if (highIterator != null) { + if (currentIterator == defaultIterator) { + orderIndex.lastDefaultKey = key; + } else if (currentIterator == highIterator) { + orderIndex.lastHighKey = key; } else { - lastDefaultKey = key; + orderIndex.lastLowKey = key; } + } else { + orderIndex.lastDefaultKey = key; } - return result; - } - - @Override - public void remove() { - throw new UnsupportedOperationException(); } + return result; } } diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBRecoverMessagesIsolatedCursorTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBRecoverMessagesIsolatedCursorTest.java new file mode 100644 index 0000000000..425fae7f28 --- /dev/null +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBRecoverMessagesIsolatedCursorTest.java @@ -0,0 +1,329 @@ +/** + * 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.store.kahadb; + +import static org.junit.Assert.assertEquals; + +import java.io.File; +import java.util.ArrayList; +import java.util.LinkedList; +import java.util.List; + +import jakarta.jms.Connection; +import jakarta.jms.JMSException; +import jakarta.jms.MessageProducer; +import jakarta.jms.Session; +import jakarta.jms.TextMessage; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.region.policy.PolicyEntry; +import org.apache.activemq.broker.region.policy.PolicyMap; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.Message; +import org.apache.activemq.command.MessageAck; +import org.apache.activemq.command.MessageId; +import org.apache.activemq.store.MessageRecoveryContext; +import org.apache.activemq.store.MessageRecoveryListener; +import org.apache.activemq.store.MessageStore; +import org.apache.activemq.store.ProxyMessageStore; +import org.apache.activemq.test.annotations.ParallelTest; +import org.apache.activemq.util.IOHelper; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.rules.TestName; + +/** + * Proves that recoverMessages(MessageRecoveryContext) with a isolated cursor + * is fully isolated from the destination's live cursor state. + * + * The MessageOrderIterator records the last visited sequence keys + * (lastDefaultKey/lastHighKey/lastLowKey) on the shared MessageOrderIndex. + * Before the fix, a isolated-cursor visit also wrote those bookmarks, and + * the next zero-entry live batch (recoverNextMessages) committed them into + * the destination cursor via stoppedIterating() — rewinding the cursor + * (duplicates from store) or jumping it forward (missed messages). + * + * Also proves that a isolated-cursor vsite does not consume rolled-back + * transactional acks that the live cursor must redeliver. + */ +@Category(ParallelTest.class) +public class KahaDBRecoverMessagesIsolatedCursorTest { + + protected BrokerService brokerService = null; + private KahaDBStore kahaDBStore = null; + + @Rule + public TestName testName = new TestName(); + + @Before + public void beforeEach() throws Exception { + kahaDBStore = createStore(true); + brokerService = createBroker(kahaDBStore); + } + + @After + public void afterEach() throws Exception { + if (brokerService != null) { + brokerService.stop(); + brokerService.waitUntilStopped(); + brokerService = null; + } + kahaDBStore = null; + } + + protected BrokerService createBroker(KahaDBStore kaha) throws Exception { + var broker = new BrokerService(); + broker.setUseJmx(false); + broker.setAdvisorySupport(false); + broker.setPersistenceAdapter(kaha); + + // keep the broker-side queue completely passive so only this test + // drives the store cursor + var policyEntry = new PolicyEntry(); + policyEntry.setExpireMessagesPeriod(0); + var policyMap = new PolicyMap(); + policyMap.setDefaultEntry(policyEntry); + broker.setDestinationPolicy(policyMap); + + broker.start(); + broker.waitUntilStarted(10_000L); + return broker; + } + + private KahaDBStore createStore(boolean delete) throws Exception { + var kaha = new KahaDBStore(); + kaha.setJournalMaxFileLength(1024 * 100); + kaha.setDirectory(new File(IOHelper.getDefaultDataDirectory(), "kahadb-isolated-cursor-tests")); + if (delete) { + kaha.deleteAllMessages(); + } + return kaha; + } + + /** + * A isolated-cursor visit between two live batches must not leave state + * behind that a later zero-entry live batch commits into the destination + * cursor. Before the fix the final recoverNextMessages() re-delivered the + * tail of the queue (duplicates from store). + */ + @Test + public void testIsolatedCursorVisitDoesNotCorruptLiveCursor() throws Exception { + var queueName = testName.getMethodName(); + sendMessages(10, queueName); + var messageStore = kahaDBStore.createQueueMessageStore(new ActiveMQQueue(queueName)); + + // live batch 1: page all 10 messages through the destination cursor + var liveBatch1 = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(10, liveBatch1); + assertEquals(10, liveBatch1.getRecoveredMessages().size()); + + // isolated-cursor visit from the head of the store (messages are + // still in the index — dispatched but unacked) + var visit = new TestMessageRecoveryListener(); + messageStore.recoverMessages(new MessageRecoveryContext.Builder() + .messageRecoveryListener(visit) + .offset(0L) + .maxMessageCountReturned(5) + .build()); + assertEquals(5, visit.getRecoveredMessages().size()); + + // live batch 2: nothing new to page in — a zero-entry batch. Before + // the fix, stoppedIterating() committed the visit bookmark here, + // rewinding the destination cursor. + var liveBatch2 = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(10, liveBatch2); + assertEquals(0, liveBatch2.getRecoveredMessages().size()); + + // live batch 3: must still be empty. Before the fix this re-delivered + // the 5 messages beyond the visit window ("duplicate from store"). + var liveBatch3 = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(10, liveBatch3); + assertEquals("live cursor re-delivered store messages after a isolated-cursor visit", + 0, liveBatch3.getRecoveredMessages().size()); + } + + /** + * A isolated-cursor visit ahead of the live cursor must not cause the + * live cursor to skip messages. Before the fix the visit bookmark was + * committed by a zero-entry live batch, jumping the cursor past messages + * that were never dispatched (stuck queue). + */ + @Test + public void testIsolatedCursorVisitDoesNotSkipLiveMessages() throws Exception { + var queueName = testName.getMethodName(); + sendMessages(10, queueName); + var messageStore = kahaDBStore.createQueueMessageStore(new ActiveMQQueue(queueName)); + + // live batch 1: page in only the first 2 messages + var liveBatch1 = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(2, liveBatch1); + assertEquals(2, liveBatch1.getRecoveredMessages().size()); + + // isolated-cursor visit across the whole store — reads sequences + // ahead of the live cursor + var visit = new TestMessageRecoveryListener(); + messageStore.recoverMessages(new MessageRecoveryContext.Builder() + .messageRecoveryListener(visit) + .offset(0L) + .maxMessageCountReturned(10) + .build()); + assertEquals(10, visit.getRecoveredMessages().size()); + + // live batch 2: must continue exactly where batch 1 stopped and + // deliver the remaining 8 messages. Before the fix a zero-entry + // batch here would have committed the visit bookmark instead, + // skipping messages 3..10 entirely. + var liveBatch2 = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(10, liveBatch2); + assertEquals("live cursor skipped messages after a isolated-cursor visit", + 8, liveBatch2.getRecoveredMessages().size()); + } + + /** + * Shared-cursor mode (useIsolatedCursor=false) intentionally advances + * the destination cursor — successive calls continue where the previous + * one stopped. This documents the contract and guards it from regressing. + */ + @Test + public void testSharedCursorModeAdvancesLiveCursor() throws Exception { + var queueName = testName.getMethodName(); + sendMessages(10, queueName); + var messageStore = kahaDBStore.createQueueMessageStore(new ActiveMQQueue(queueName)); + + var firstPage = new TestMessageRecoveryListener(); + messageStore.recoverMessages(new MessageRecoveryContext.Builder() + .messageRecoveryListener(firstPage) + .useIsolatedCursor(false) + .maxMessageCountReturned(5) + .build()); + assertEquals(5, firstPage.getRecoveredMessages().size()); + assertEquals(0, firstPage.getRecoveredMessages().get(0).getProperty("index")); + + var secondPage = new TestMessageRecoveryListener(); + messageStore.recoverMessages(new MessageRecoveryContext.Builder() + .messageRecoveryListener(secondPage) + .useIsolatedCursor(false) + .maxMessageCountReturned(5) + .build()); + assertEquals(5, secondPage.getRecoveredMessages().size()); + assertEquals(5, secondPage.getRecoveredMessages().get(0).getProperty("index")); + + // the shared cursor is now at the tail — the live batch sees nothing + var liveBatch = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(10, liveBatch); + assertEquals(0, liveBatch.getRecoveredMessages().size()); + } + + /** + * Rolled-back transactional acks are queued for redelivery through the + * live cursor. A isolated-cursor visit must not consume them: before + * the fix the visit both received the rolled-back message (duplicating + * it in its own scan results) and permanently removed it from the + * redelivery map, so the live cursor never redelivered it. + */ + @Test + public void testIsolatedCursorVisitDoesNotConsumeRolledBackAcks() throws Exception { + var queueName = testName.getMethodName(); + sendMessages(3, queueName); + var messageStore = kahaDBStore.createQueueMessageStore(new ActiveMQQueue(queueName)); + + // live batch 1: page all 3 messages through the destination cursor + var liveBatch1 = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(3, liveBatch1); + assertEquals(3, liveBatch1.getRecoveredMessages().size()); + MessageId rolledBackId = liveBatch1.getRecoveredMessages().get(1).getMessageId(); + + // simulate an XA prepare + rollback outcome for the second message: + // the store queues it for redelivery via the live cursor + KahaDBStore.KahaDBMessageStore kahaMessageStore = + (KahaDBStore.KahaDBMessageStore) ((ProxyMessageStore) messageStore).getDelegate(); + var rolledBackAck = new MessageAck(); + rolledBackAck.setLastMessageId(rolledBackId); + var acks = new ArrayList<MessageAck>(); + acks.add(rolledBackAck); + kahaMessageStore.trackRecoveredAcks(acks); + kahaMessageStore.forgetRecoveredAcks(acks, true); + + // isolated-cursor visit: must see exactly the 3 messages in the + // index — no rolled-back-ack replay mixed into visit results + var visit = new TestMessageRecoveryListener(); + messageStore.recoverMessages(new MessageRecoveryContext.Builder() + .messageRecoveryListener(visit) + .offset(0L) + .maxMessageCountReturned(10) + .build()); + assertEquals("isolated-cursor visit consumed rolled-back ack redeliveries", + 3, visit.getRecoveredMessages().size()); + + // the live cursor must still redeliver the rolled-back message + var liveBatch2 = new TestMessageRecoveryListener(); + messageStore.recoverNextMessages(10, liveBatch2); + assertEquals("rolled-back ack was not redelivered through the live cursor", + 1, liveBatch2.getRecoveredMessages().size()); + assertEquals(rolledBackId, liveBatch2.getRecoveredMessages().get(0).getMessageId()); + } + + private void sendMessages(int count, String queueName) throws JMSException { + var cf = new ActiveMQConnectionFactory("vm://localhost"); + cf.setWatchTopicAdvisories(false); + + try (var connection = cf.createConnection(); + var session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var producer = session.createProducer(new ActiveMQQueue(queueName))) { + + for (int i = 0; i < count; i++) { + var textMessage = session.createTextMessage("message:" + i); + textMessage.setIntProperty("index", i); + producer.send(textMessage); + } + } + } + + static class TestMessageRecoveryListener implements MessageRecoveryListener { + + final List<MessageId> recoveredMessageIds = new LinkedList<>(); + final List<Message> recoveredMessages = new LinkedList<>(); + + @Override + public boolean hasSpace() { + return true; + } + + @Override + public boolean isDuplicate(MessageId messageId) { + return recoveredMessageIds.contains(messageId); + } + + @Override + public boolean recoverMessage(Message message) throws Exception { + return recoveredMessages.add(message); + } + + @Override + public boolean recoverMessageReference(MessageId messageId) throws Exception { + return recoveredMessageIds.add(messageId); + } + + public List<Message> getRecoveredMessages() { + return recoveredMessages; + } + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected] For further information, visit: https://activemq.apache.org/contact
