Skip to content

[fix] Fix ManagedLedgerImpl's pendingAddEntries leak. #25240

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
zhaizhibo wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[fix] Fix ManagedLedgerImpl's pendingAddEntries leak. #25240

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
eb9902e
Fix ManagedLedgerImpl's pendingAddEntries leak.
Feb 11, 2026
1363918
Fix comments.
Apr 30, 2026
b86304a
Fix test compilation: use string concatenation for logger instead of …
May 1, 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
69 changes: 51 additions & 18 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 @@ -1652,6 +1652,7 @@ public synchronized void asyncClose(final CloseCallback callback, final Object c

factory.close(this);
STATE_UPDATER.set(this, State.Closed);
clearPendingAddEntries(new ManagedLedgerAlreadyClosedException("Managed ledger is closed"));
cancelScheduledTasks();

LedgerHandle lh = currentLedger;
Expand Down Expand Up @@ -1746,19 +1747,39 @@ public void operationComplete(Void v, Stat stat) {
log.debug().attr("version", stat).log("Updating of ledgers list after create complete");
ledgersStat = stat;
synchronized (ManagedLedgerImpl.this) {
LedgerHandle originalCurrentLedger = currentLedger;
ledgers.put(lh.getId(), newLedger);
currentLedger = lh;
currentLedgerTimeoutTriggered = new AtomicBoolean();
currentLedgerEntries = 0;
currentLedgerSize = 0;
updateLedgersIdsComplete(originalCurrentLedger);
mbean.addLedgerSwitchLatencySample(System.currentTimeMillis()
- lastLedgerCreationInitiationTimestamp, TimeUnit.MILLISECONDS);
// May need to update the cursor position
maybeUpdateCursorBeforeTrimmingConsumedLedger();
try {
State state = STATE_UPDATER.get(ManagedLedgerImpl.this);
if (state == State.Closed || state.isFenced()) {
log.debug()
.attr("name", name)
.log("skip ledger update after create complete ledger is closed or fenced");
lh.closeAsync().exceptionally(e -> {
if (e != null) {
log.error()
.attr("ledgerName", name)
.attr("ledgerId", lh.getId())
.attr("error", e.getMessage())
.log("Failed to close ledger");
}
return null;
});
} else {
LedgerHandle originalCurrentLedger = currentLedger;
ledgers.put(lh.getId(), newLedger);
currentLedger = lh;
currentLedgerTimeoutTriggered = new AtomicBoolean();
currentLedgerEntries = 0;
currentLedgerSize = 0;
updateLedgersIdsComplete(originalCurrentLedger);
mbean.addLedgerSwitchLatencySample(System.currentTimeMillis()
- lastLedgerCreationInitiationTimestamp, TimeUnit.MILLISECONDS);
// May need to update the cursor position
maybeUpdateCursorBeforeTrimmingConsumedLedger();
}
} finally {
metadataMutex.unlock();
}
}
metadataMutex.unlock();
}

@Override
Expand Down Expand Up @@ -1927,6 +1948,9 @@ synchronized void ledgerClosed(final LedgerHandle lh, Long lastAddConfirmed) {
// The managed ledger was closed during the write operation
clearPendingAddEntries(new ManagedLedgerAlreadyClosedException("Managed ledger was already closed"));
return;
} else if (state.isFenced()) {
clearPendingAddEntries(new ManagedLedgerFencedException("Managed ledger is fenced"));
return;
} else {
// In case we get multiple write errors for different outstanding write request, we should close the ledger
// just once
Expand Down Expand Up @@ -2052,7 +2076,7 @@ public ManagedLedgerInterceptor getManagedLedgerInterceptor() {
return managedLedgerInterceptor;
}

void clearPendingAddEntries(ManagedLedgerException e) {
synchronized void clearPendingAddEntries(ManagedLedgerException e) {
while (!pendingAddEntries.isEmpty()) {
OpAddEntry op = pendingAddEntries.poll();
op.failed(e);
Expand Down Expand Up @@ -4327,7 +4351,8 @@ public boolean isCursorActive(ManagedCursor cursor) {
return activeCursors.get(cursor.getName()) != null;
}

private boolean currentLedgerIsFull() {
@VisibleForTesting
protected boolean currentLedgerIsFull() {
if (!factory.isMetadataServiceAvailable()) {
// We don't want to trigger metadata operations if we already know that we're currently disconnected
return false;
Expand Down Expand Up @@ -4416,13 +4441,21 @@ private void checkManagedLedgerIsOpen() throws ManagedLedgerException {

@VisibleForTesting
public synchronized void setFenced() {
log.info("Moving to Fenced state");
STATE_UPDATER.set(this, State.Fenced);
log.info().attr("ledgerName", name).log("Moving to Fenced state");
if (STATE_UPDATER.get(this) != State.Fenced) {
STATE_UPDATER.set(this, State.Fenced);
clearPendingAddEntries(new ManagedLedgerFencedException("ManagedLedger "
+ name + " is fenced"));
}
}

synchronized void setFencedForDeletion() {
log.info("Moving to FencedForDeletion state");
STATE_UPDATER.set(this, State.FencedForDeletion);
log.info().attr("ledgerName", name).log("Moving to FencedForDeletion state");
Comment thread
zhaizhibo marked this conversation as resolved.
if (STATE_UPDATER.get(this) != State.FencedForDeletion) {
STATE_UPDATER.set(this, State.FencedForDeletion);
clearPendingAddEntries(new ManagedLedgerFencedException("ManagedLedger "
+ name + " is fenced"));
}
}

MetaStore getStore() {
Expand Down
88 changes: 88 additions & 0 deletions managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -764,6 +764,94 @@ public void invalidReadEntriesArg2() throws Exception {
fail("Should have thrown an exception in the above line");
}

@Test(timeOut = 30000)
public void testCloseManagedLedgerAfterRollover() throws Exception {
ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig();
config.setMaxCacheSize(0);
@Cleanup("shutdown")
ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, config);
Comment thread
zhaizhibo marked this conversation as resolved.
ManagedLedgerImpl realLedger = (ManagedLedgerImpl) factory.open("my_test_ledger");
ManagedLedgerImpl ledger = Mockito.spy(realLedger);
AtomicBoolean onlyOnce = new AtomicBoolean(false);
when(ledger.currentLedgerIsFull()).thenAnswer(invocation -> onlyOnce.compareAndSet(false, true));
OpAddEntry realOp = OpAddEntry.createNoRetainBuffer(ledger,
ByteBufAllocator.DEFAULT.buffer(128), null, null, new AtomicBoolean());
OpAddEntry op = spy(realOp);
CountDownLatch createLatch = new CountDownLatch(1);
CountDownLatch closeLatch = new CountDownLatch(1);
doAnswer(invocationOnMock -> {
// Simulate that before the rollover is completed, new write requests arrive,
// and after these write requests are added to pendingAddEntries, the ledger is closed.
log.info("before add, ledger state: " + ledger.getState());
for (int i = 0; i < 10; ++i) {
ledger.internalAsyncAddEntry(OpAddEntry.createNoRetainBuffer(ledger,
ByteBufAllocator.DEFAULT.buffer(128), null, null, new AtomicBoolean()));
}
ledger.asyncClose(new CloseCallback() {
@Override
public void closeComplete(Object ctx) {
log.info("closeComplete finished, ledger state: " + ledger.getState());
closeLatch.countDown();
}

@Override
public void closeFailed(ManagedLedgerException exception, Object ctx) {
log.info("closeFailed, ex: " + exception.getMessage() + ", state: " + ledger.getState());
closeLatch.countDown();
}
}, null);
log.info("after add, ledger state: " + ledger.getState());
return invocationOnMock.callRealMethod();
}).when(ledger).asyncCreateLedger(any(), any(), any(), any(), any());
doAnswer(invocationOnMock -> {
Object o = invocationOnMock.callRealMethod();
log.info("createComplete finished, state: " + ledger.getState());
ledger.executor.execute(createLatch::countDown);
return o;
}).when(ledger).createComplete(anyInt(), any(), any());
ledger.internalAsyncAddEntry(op);
createLatch.await();
closeLatch.await();
Assert.assertEquals(ledger.pendingAddEntries.size(), 0);
}

@Test(timeOut = 20000)
public void testFencedManagedLedgerAfterAdd() throws Exception {
@Cleanup("shutdown")
ManagedLedgerFactory factory1 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
ManagedLedgerImpl realLedger = (ManagedLedgerImpl) factory1.open("my_test_ledger");
ManagedLedgerImpl ledger = spy(realLedger);

int sendNum = 10;
CountDownLatch sendLatch = new CountDownLatch(sendNum);
CountDownLatch fencedLatch = new CountDownLatch(1);
doAnswer(invocationOnMock -> {
stopBookKeeper();
stopMetadataStore();
for (int i = 0; i < sendNum; ++i) {
Comment thread
zhaizhibo marked this conversation as resolved.
ledger.internalAsyncAddEntry(OpAddEntry.createNoRetainBuffer(ledger,
ByteBufAllocator.DEFAULT.buffer(128), new AddEntryCallback() {
@Override
public void addComplete(Position position, ByteBuf entryData, Object ctx) {
sendLatch.countDown();
}

@Override
public void addFailed(ManagedLedgerException exception, Object ctx) {
sendLatch.countDown();
}
}, null, new AtomicBoolean()));
}
Object o = invocationOnMock.callRealMethod();
fencedLatch.countDown();
return o;
}).when(ledger).setFenced();
ledger.setFenced();
fencedLatch.await();
sendLatch.await();
assertEquals(ledger.pendingAddEntries.size(), 0);
}

@Test(timeOut = 20000)
public void deleteAndReopen() throws Exception {
ManagedLedger ledger = factory.open("my_test_ledger");
Expand Down
6 changes: 5 additions & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -764,17 +764,21 @@ public synchronized void addFailed(ManagedLedgerException exception, Object ctx)
Instead, we will rely on the service unit state channel's bundle(topic) transfer protocol.
At the end of the transfer protocol, at Owned state, the source broker should close the topic properly.
*/
PublishContext callback = (PublishContext) ctx;
if (transferring) {
log.debug()
.exception(exception)
.log("Failed to persist msg in store while transferring");
callback.completed(new TopicClosedException(exception), -1, -1);
decrementPendingWriteOpsAndCheck();
return;
}

PublishContext callback = (PublishContext) ctx;
if (exception instanceof ManagedLedgerFencedException) {
// If the managed ledger has been fenced, we cannot continue using it. We need to close and reopen
close();
callback.completed(new TopicFencedException(exception.getMessage()), -1, -1);
decrementPendingWriteOpsAndCheck();
} else {
// fence topic when failed to write a message to BK
fence();
Expand Down