Skip to content

[#2230] KahaDB dedicated cursor iterator #2231

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mattrpav wants to merge 5 commits into
base: main
Choose a base branch
from
Open

[#2230] KahaDB dedicated cursor iterator #2231

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a539e7d
[#2230] Unit test demonstrating dedicatedCursor and current dedicated…
mattrpav Jul 18, 2026
7a4d037
[#2230] KahaDB optionally use a isolatedIterator with isolatedCursor
mattrpav Jul 18, 2026
68eea73
[#] Refactor to have MessageOrderIterator extend IsolatedMessageOrder…
mattrpav Jul 20, 2026
77de343
[#] Fix for recover() and recoverExpired() to use IsolatedMessageOrde…
mattrpav Jul 20, 2026
cf8c34c
[#] Add iteratorIsolated(..) overloads to the MessageOrderIndex
mattrpav Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 20 additions & 10 deletions activemq-broker/src/main/java/org/apache/activemq/store/MessageRecoveryContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,10 @@
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;
Expand All @@ -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");
}
Expand All @@ -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() {
Expand Down Expand Up @@ -133,26 +138,31 @@ public int getRecoveredCount() {

@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 + "]";
}

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;
Expand All @@ -179,7 +189,7 @@ public Builder messageRecoveryListener(final MessageRecoveryListener messageReco
}

public MessageRecoveryContext build() {
return new MessageRecoveryContext(messageRecoveryListener, startMessageId, endMessageId, offset, maxMessageCountReturned, useDedicatedCursor);
return new MessageRecoveryContext(messageRecoveryListener, startMessageId, endMessageId, offset, maxMessageCountReturned, useIsolatedCursor);
}
}
}
Expand Down
10 changes: 5 additions & 5 deletions activemq-broker/src/test/java/org/apache/activemq/store/MessageRecoveryContextTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,17 +39,17 @@ public void testConfigOffset() {
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);
Expand All @@ -58,7 +58,7 @@ public void testConfigOffsetNoDedicatedCursor() {
assertNotNull(messageRecoveryContext.getMessageRecoveryListener());
assertEquals(Long.valueOf(10_000l), Long.valueOf(messageRecoveryContext.getOffset()));
assertNull(messageRecoveryContext.getStartMessageId());
assertFalse(messageRecoveryContext.isUseDedicatedCursor());
assertFalse(messageRecoveryContext.isUseIsolatedCursor());
}

@Test
Expand All @@ -77,7 +77,7 @@ public void testConfigStartEndMsgId() {
assertNotNull(messageRecoveryContext.getMessageRecoveryListener());
assertNull(messageRecoveryContext.getOffset());
assertEquals("ID-start-12", messageRecoveryContext.getStartMessageId());
assertTrue(messageRecoveryContext.isUseDedicatedCursor());
assertTrue(messageRecoveryContext.isUseIsolatedCursor());
}

@Test(expected = IllegalArgumentException.class)
Expand Down
43 changes: 34 additions & 9 deletions activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -656,9 +656,8 @@ public void recover(final MessageRecoveryListener listener) throws Exception {
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());
Expand Down Expand Up @@ -703,10 +702,28 @@ public void recoverNextMessages(final int maxReturned, final MessageRecoveryList
}
}

/**
* 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);
Expand Down Expand Up @@ -759,10 +776,19 @@ public void recoverMessages(final MessageRecoveryContext messageRecoveryContext)
}

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();
Expand All @@ -781,7 +807,7 @@ public void recoverMessages(final MessageRecoveryContext messageRecoveryContext)
}

// [AMQ-9773] The sd.orderIndex uses the destination's cursor
if(!messageRecoveryContext.isUseDedicatedCursor()) {
if(!messageRecoveryContext.isUseIsolatedCursor()) {
sd.orderIndex.stoppedIterating();
}
});
Expand Down Expand Up @@ -1280,7 +1306,6 @@ public Map<SubscriptionKey,List<Message>> recoverExpired(Set<SubscriptionKey> su
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<>();
Expand All @@ -1297,7 +1322,7 @@ public Map<SubscriptionKey,List<Message>> recoverExpired(Set<SubscriptionKey> su
// 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());
Expand Down
Loading