Skip to content

[fix][ml] Preserve cursor properties when recovering from an unreadable cursor ledger #26512

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

Merged
lhotari merged 3 commits into from
Sep 10, 2026
Merged

[fix][ml] Preserve cursor properties when recovering from an unreadable cursor ledger #26512

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c4a18b4
[fix][ml] Preserve cursor properties when recovering from an unreadab…
dao-jun Sep 9, 2026
0a8610c
address review comments
dao-jun Sep 10, 2026
abcd0d8
[fix][broker] Capture the compacted topic context once in readLastEnt…
dao-jun Sep 10, 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
45 changes: 24 additions & 21 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -583,15 +583,8 @@ public void operationComplete(ManagedCursorInfo info, Stat stat) {
info::getIndividualDeletedMessageAt);
}

Map<String, Long> recoveredProperties = Collections.emptyMap();
if (info.getPropertiesCount() > 0) {
// Recover properties map
recoveredProperties = new HashMap<>();
for (int i = 0; i < info.getPropertiesCount(); i++) {
LongProperty property = info.getPropertyAt(i);
recoveredProperties.put(property.getName(), property.getValue());
}
}
Map<String, Long> recoveredProperties =
recoverProperties(info.getPropertiesCount(), info::getPropertyAt);

recoveredCursor(recoveredPosition, recoveredProperties, recoveredCursorProperties, null);
callback.operationComplete();
Expand All @@ -614,6 +607,10 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
// a new ledger and write the position into it
ledger.mbean.startCursorLedgerOpenOp();
long ledgerId = info.getCursorsLedgerId();
// If the cursor ledger cannot be read, the cursor is rolled back to the position snapshotted in
// ManagedCursorInfo. The properties saved alongside that snapshot must be carried over: initialize()
// persists whatever map it receives, so passing an empty map would durably wipe them out.
Map<String, Long> rollbackProperties = recoverProperties(info.getPropertiesCount(), info::getPropertyAt);
OpenCallback openCallback = (rc, lh, ctx) -> {
log.info().attr("ledgerId", ledgerId).attr("rc", rc).log("Opened ledger");
if (isBkErrorNotRecoverable(rc) || (rc != BKException.Code.OK && ledgerForceRecovery)) {
Expand All @@ -622,7 +619,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
.attr("errorMessage", BKException.getMessage(rc))
.log("Error opening metadata ledger");
// Rewind to the oldest entry available
initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
initialize(getRollbackPosition(info), rollbackProperties, cursorProperties, callback);
Comment thread
dao-jun marked this conversation as resolved.
return;
} else if (rc != BKException.Code.OK) {
log.warn()
Expand All @@ -639,7 +636,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
if (lastEntryInLedger < 0) {
log.warn().attr("ledgerId", ledgerId).log("Error reading from metadata ledger: no entries in ledger");
// Rewind to last cursor snapshot available
initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
initialize(getRollbackPosition(info), rollbackProperties, cursorProperties, callback);
return;
}

Expand All @@ -651,7 +648,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
.attr("errorMessage", BKException.getMessage(rc1))
.log("Error reading from metadata ledger");
// Rewind to the oldest entry available
initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
initialize(getRollbackPosition(info), rollbackProperties, cursorProperties, callback);
return;
} else if (rc1 != BKException.Code.OK) {
log.warn()
Expand All @@ -673,15 +670,8 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
return;
}

Map<String, Long> recoveredProperties = Collections.emptyMap();
if (positionInfo.getPropertiesCount() > 0) {
// Recover properties map
recoveredProperties = new HashMap<>();
for (int i = 0; i < positionInfo.getPropertiesCount(); i++) {
LongProperty property = positionInfo.getPropertyAt(i);
recoveredProperties.put(property.getName(), property.getValue());
}
}
Map<String, Long> recoveredProperties =
recoverProperties(positionInfo.getPropertiesCount(), positionInfo::getPropertyAt);

Position position = PositionFactory.create(positionInfo.getLedgerId(), positionInfo.getEntryId());
recoverIndividualDeletedMessages(positionInfo);
Expand Down Expand Up @@ -749,6 +739,19 @@ private List<LongListMap> buildLongPropertiesMap(Map<Long, long[]> properties) {
return longListMap;
}

private static Map<String, Long> recoverProperties(int count, IntFunction<LongProperty> accessor) {
Map<String, Long> properties = Collections.emptyMap();
if (count > 0) {
// Recover properties map
properties = new HashMap<>();
for (int i = 0; i < count; i++) {
LongProperty property = accessor.apply(i);
properties.put(property.getName(), property.getValue());
}
}
return properties;
}

@VisibleForTesting
void recoverIndividualDeletedMessages(int count, IntFunction<MessageRange> accessor) {
lock.writeLock().lock();
Expand Down
Loading
Loading