This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/main by this push:
new d74a6a179d [#2230] KahaDB optionally use a isolatedIterator with
isolatedCursor
d74a6a179d is described below
commit d74a6a179d88a20df767252e241d0bbdf4d744e7
Author: Matt Pavlovich <[email protected]>
AuthorDate: Sat Jul 18 09:02:37 2026 -0500
[#2230] KahaDB optionally use a isolatedIterator with isolatedCursor
---
.../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 fb3eb7a254..68cfc65708 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 08751b9c3b..5cbdc67575 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
@@ -3848,6 +3848,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);
}
@@ -3856,6 +3862,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;
}
@@ -3888,100 +3902,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
