Skip to content

[fix][ml] Make managed ledger properties updates atomic #26569

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
void-ptr974 wants to merge 2 commits into
base: master
Choose a base branch
from
+325 −25
Open

[fix][ml] Make managed ledger properties updates atomic #26569

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
216c237
[fix][ml] Make managed ledger properties updates atomic
void-ptr974 Sep 13, 2026
38b196b
[fix][ml] Preserve migration marker during property updates
void-ptr974 Sep 15, 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
3 changes: 2 additions & 1 deletion managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -620,7 +620,8 @@ default void unfenceForInterceptorException() {
void readyToCreateNewLedger();

/**
* Returns managed-ledger's properties.
* Returns a snapshot of the managed-ledger's properties.
* Changes made to the returned map are not applied to the managed ledger; use the property update methods instead.
*
* @return key-values of properties
*/
Expand Down
96 changes: 72 additions & 24 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,7 +172,7 @@ public Logger getLogger() {
private final AtomicReference<Position> cacheEvictionPosition = new AtomicReference<>();

protected ManagedLedgerConfig config;
protected Map<String, String> propertiesMap;
protected volatile Map<String, String> propertiesMap;
protected final MetaStore store;

final ConcurrentLongHashMap<CompletableFuture<ReadHandle>> ledgerCache =
Expand Down Expand Up @@ -445,17 +445,17 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
ledgers.put(ls.getLedgerId(), ls);
}

if (mlInfo.getPropertiesCount() > 0) {
propertiesMap = new HashMap<>();
for (int i = 0; i < mlInfo.getPropertiesCount(); i++) {
KeyValue property = mlInfo.getPropertyAt(i);
propertiesMap.put(property.getKey(), property.getValue());
}
Map<String, String> loadedProperties = new ConcurrentHashMap<>();
for (int i = 0; i < mlInfo.getPropertiesCount(); i++) {
KeyValue property = mlInfo.getPropertyAt(i);
loadedProperties.put(property.getKey(), property.getValue());
}
migrated = mlInfo.hasTerminatedPosition() && propertiesMap.containsKey(MIGRATION_STATE_PROPERTY);
migrated = mlInfo.hasTerminatedPosition()
&& loadedProperties.containsKey(MIGRATION_STATE_PROPERTY);
if (managedLedgerInterceptor != null) {
managedLedgerInterceptor.onManagedLedgerPropertiesInitialize(propertiesMap);
managedLedgerInterceptor.onManagedLedgerPropertiesInitialize(loadedProperties);
}
propertiesMap = loadedProperties;

// Last ledger stat may be zeroed, we must update it
if (!ledgers.isEmpty()) {
Expand Down Expand Up @@ -1398,8 +1398,22 @@ private long consumedLedgerSize(long ledgerSize, long ledgerEntries, long consum
}

public CompletableFuture<Position> asyncMigrate() {
propertiesMap.put(MIGRATION_STATE_PROPERTY, Boolean.TRUE.toString());
CompletableFuture<Position> result = new CompletableFuture<>();
asyncSetProperty(MIGRATION_STATE_PROPERTY, Boolean.TRUE.toString(), new UpdatePropertiesCallback() {
@Override
public void updatePropertiesComplete(Map<String, String> properties, Object ctx) {
terminateForMigration(result);

@Denovo1998 Denovo1998 Sep 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we also serialize the termination metadata update with metadataMutex?

asyncSetProperty() releases the mutex before its callback executes, while asyncTerminate() later updates the same ManagedLedgerInfo metadata (including ledgersStat) without acquiring metadataMutex.

If another property update begins while the BookKeeper close is still pending, both operations may call asyncUpdateLedgerIds() with the same metadata version. The second to complete could then encounter a BadVersion error, fencing or failing the managed ledger.

This appears to be the same type of race condition that metadataMutex was introduced to prevent in #7161/#7357.

Could we add a regression test that triggers another property update after the migration-marker callback but before the BookKeeper close callback is invoked? I think either the termination metadata write should also be protected by metadataMutex, or migration should persist both the marker and the terminated state in a single serialized metadata update.

}

@Override
public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx) {
result.completeExceptionally(exception);
}
}, null);
return result;
}

private void terminateForMigration(CompletableFuture<Position> result) {
asyncTerminate(new TerminateCallback() {

@Override
Expand All @@ -1415,7 +1429,6 @@ public void terminateFailed(ManagedLedgerException exception, Object ctx) {
result.completeExceptionally(exception);
}
}, null);
return result;
}

@Override
Expand Down Expand Up @@ -4462,21 +4475,31 @@ private ManagedLedgerInfo getManagedLedgerInfo(LedgerInfo newLedger) {
return buildManagedLedgerInfo(mlInfo);
}
private ManagedLedgerInfo buildManagedLedgerInfo(Map<Long, LedgerInfo> ledgers) {
return buildManagedLedgerInfo(ledgers, propertiesMap);
}

private ManagedLedgerInfo buildManagedLedgerInfo(Map<Long, LedgerInfo> ledgers,
Map<String, String> properties) {
ManagedLedgerInfo mlInfo = new ManagedLedgerInfo();
mlInfo.addAllLedgerInfos(ledgers.values());
return buildManagedLedgerInfo(mlInfo);
return buildManagedLedgerInfo(mlInfo, properties);
}

private ManagedLedgerInfo buildManagedLedgerInfo(ManagedLedgerInfo mlInfo) {
return buildManagedLedgerInfo(mlInfo, propertiesMap);
}

private ManagedLedgerInfo buildManagedLedgerInfo(ManagedLedgerInfo mlInfo,
Map<String, String> properties) {
if (state == State.Terminated) {
mlInfo.setTerminatedPosition()
.setLedgerId(lastConfirmedEntry.getLedgerId())
.setEntryId(lastConfirmedEntry.getEntryId());
}
if (managedLedgerInterceptor != null) {
managedLedgerInterceptor.onUpdateManagedLedgerInfo(propertiesMap);
managedLedgerInterceptor.onUpdateManagedLedgerInfo(properties);
}
for (Map.Entry<String, String> property : propertiesMap.entrySet()) {
for (Map.Entry<String, String> property : properties.entrySet()) {
mlInfo.addProperty().setKey(property.getKey()).setValue(property.getValue());
}

Expand Down Expand Up @@ -4809,7 +4832,7 @@ public long getLastOffloadedFailureTimestamp() {

@Override
public Map<String, String> getProperties() {
return propertiesMap;
return new HashMap<>(propertiesMap);
}

@Override
Expand Down Expand Up @@ -4838,13 +4861,13 @@ public void asyncDeleteProperty(String key, final UpdatePropertiesCallback callb

@Override
public void setProperties(Map<String, String> properties) throws InterruptedException, ManagedLedgerException {
updateProperties(properties, false, null);
updateProperties(new HashMap<>(properties), false, null);
}

@Override
public void asyncSetProperties(Map<String, String> properties, final UpdatePropertiesCallback callback,
Object ctx) {
asyncUpdateProperties(properties, false, null, callback, ctx);
asyncUpdateProperties(new HashMap<>(properties), false, null, callback, ctx);
}

private void updateProperties(Map<String, String> properties, boolean isDelete,
Expand Down Expand Up @@ -4885,25 +4908,50 @@ private void asyncUpdateProperties(Map<String, String> properties, boolean isDel
callback, ctx), 100, TimeUnit.MILLISECONDS);
return;
}
Map<String, String> updatedProperties = new HashMap<>(propertiesMap);
if (isDelete) {
propertiesMap.remove(deleteKey);
updatedProperties.remove(deleteKey);
} else {
propertiesMap.putAll(properties);
updatedProperties.putAll(properties);
}

final ManagedLedgerInfo managedLedgerInfo;
final Map<String, String> propertiesSnapshot;
try {
managedLedgerInfo = buildManagedLedgerInfo(ledgers, updatedProperties);
propertiesSnapshot = new ConcurrentHashMap<>(updatedProperties);
} catch (Throwable t) {
metadataMutex.unlock();
callback.updatePropertiesFailed(ManagedLedgerException.getManagedLedgerException(t), ctx);
return;
}
store.asyncUpdateLedgerIds(name, getManagedLedgerInfo(), ledgersStat, new MetaStoreCallback<Void>() {

store.asyncUpdateLedgerIds(name, managedLedgerInfo, ledgersStat, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Stat version) {
ledgersStat = version;
callback.updatePropertiesComplete(propertiesMap, ctx);
propertiesMap = propertiesSnapshot;

@lhotari lhotari Sep 13, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[BUG] Publishing this snapshot can erase a concurrent migration marker

asyncMigrate() writes migrated=true directly to the currently published map at ManagedLedgerImpl.java:1400-1403 without taking metadataMutex. If a property update has already captured propertiesSnapshot, migration can add the marker while its BookKeeper close is pending, and this callback can then replace the map with the older snapshot. The close callback subsequently serializes that replacement using the updated ledgersStat, so both operations can succeed while the terminated metadata lacks the marker; reopening the ledger then reports isMigrated() as false. Please serialize the internal migration write with snapshot publication and add a deterministic regression test that gates the property metadata callback and BookKeeper close callback in this order.

@void-ptr974 void-ptr974 Sep 15, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, good catch. Fixed in 38b196b.

asyncMigrate() now writes the migration marker through asyncSetProperty() and starts termination only after the property update succeeds. This serializes it with other property snapshot updates under metadataMutex.

I added a deterministic regression test that releases the pending property callback before the gated BookKeeper close callback, then reopens the ledger and verifies the migration marker is preserved. It fails on the previous revision and passes with the fix. quickCheck and the focused tests also pass.

metadataMutex.unlock();
try {
callback.updatePropertiesComplete(new HashMap<>(propertiesSnapshot), ctx);
} catch (Throwable t) {
log.error().exception(t).log("Managed ledger properties callback failed");
}
}

@Override
public void operationFailed(MetaStoreException e) {
log.error().exception(e).log("Update managedLedger's properties failed");
handleBadVersion(e);
callback.updatePropertiesFailed(e, ctx);
metadataMutex.unlock();
try {
handleBadVersion(e);
} finally {
metadataMutex.unlock();
}
try {
callback.updatePropertiesFailed(e, ctx);
} catch (Throwable t) {
log.error().exception(t).log("Managed ledger properties callback failed");
}
}
});
}
Expand Down
Loading