You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "void-ptr974 (via GitHub)" <gi...@apache.org> on 2026/10/02 14:58:32 UTC
[PR] [fix][ml] Close cached handles when a read-only managed ledger closes [pulsar]
void-ptr974 opened a new pull request, #26816:
URL: https://github.com/apache/pulsar/pull/26816
Main Issue: #26815
### Motivation
A read-only managed ledger has no current write handle. Its close path therefore skips the write-handle callback that normally releases cached read handles, leaving their metadata listeners registered.
### Modifications
Release the read-only managed ledger's cached handles when it closes, including handles whose open is still pending. Reuse the existing cache invalidation cleanup without changing writable managed-ledger behavior.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
Opened as a draft while full CI validation is pending ([fork CI](https://github.com/void-ptr974/pulsar/actions/runs/37022299384)).
Added `ReadOnlyManagedLedgerHandleCloseTest` covering normal close, fenced close, and pending opens.
Passed 11 focused tests and `./gradlew quickCheck`. All three new cases fail on the unpatched code:
```shell
./gradlew :managed-ledger:test --tests '*ReadOnlyManagedLedgerHandleCloseTest' --tests '*ReadOnlyCursorTest' -PtestRetryCount=0
```
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Fix cached read handle leaks when closing read-only managed ledgers [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26816:
URL: https://github.com/apache/pulsar/pull/26816#discussion_r4231010037
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java:
##########
@@ -156,6 +160,29 @@ public ReadOnlyCursor createReadOnlyCursor(Position startPosition) {
return new ReadOnlyCursorImpl(bookKeeper, this, startPosition, "read-only-cursor");
}
+ @Override
+ public synchronized void asyncClose(AsyncCallbacks.CloseCallback callback, Object ctx) {
+ closingOrClosed = true;
+ // There is no current write handle, so the parent's write-close callback cannot release this cache.
+ ledgerCache.forEach((ledgerId, readHandle) -> invalidateReadHandle(ledgerId));
+ super.asyncClose(callback, ctx);
Review Comment:
I think there's still a lifecycle gap in the actual factory path.
`ManagedLedgerFactoryImpl.asyncOpenReadOnlyManagedLedger()` returns a `ReadOnlyManagedLedgerImplWrapper`, but neither the public `ReadOnlyManagedLedger` interface nor the wrapper exposes a close method.
The read-only cursor path looks similar: `ReadOnlyCursorImpl.asyncClose()` only closes the cursor, not the backing managed ledger.
So these callers never reach this new `asyncClose()` implementation, and the cached handles can still be left open. One example is `SnapshotSegmentAbortedTxnProcessorImpl#readSegmentEntries()`.
Could we also wire up the ownership/close path for the factory-facing APIs? A regression test through the factory API would help catch this.
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerHandleCloseTest.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import static org.apache.bookkeeper.mledger.util.ManagedLedgerTestUtil.defaultConfig;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.OpenBuilder;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerAlreadyClosedException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.awaitility.Awaitility;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(timeOut = 30000)
+public class ReadOnlyManagedLedgerHandleCloseTest extends MockedBookKeeperTestCase {
+ private final List<LedgerHandle> openedHandles = new CopyOnWriteArrayList<>();
+ private CompletableFuture<Void> openCompletion;
+
+ @Override
+ protected void setUpTestCase() throws Exception {
+ openedHandles.clear();
+ openCompletion = CompletableFuture.completedFuture(null);
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ bkc = spy(bkc);
+ doAnswer(invocation -> trackOpens((OpenBuilder) invocation.callRealMethod()))
+ .when(bkc).newOpenLedgerOp();
+ factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ }
+
+ @Override
+ protected void cleanUpTestCase() throws Exception {
+ openCompletion.complete(null);
+ // Initialization's temporary handle is covered by a separate fix.
+ for (LedgerHandle handle : openedHandles) {
+ doCallRealMethod().when(handle).closeAsync();
+ handle.closeAsync().get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private OpenBuilder trackOpens(OpenBuilder builder) {
+ OpenBuilder trackedBuilder = spy(builder);
+ doAnswer(invocation -> ((CompletableFuture<ReadHandle>) invocation.callRealMethod()).thenCompose(handle -> {
+ LedgerHandle trackedHandle = spy((LedgerHandle) handle);
+ openedHandles.add(trackedHandle);
+ return openCompletion.thenApply(ignored -> trackedHandle);
+ })).when(trackedBuilder).execute();
+ return trackedBuilder;
+ }
+
+ @DataProvider(name = "closeStates")
+ public Object[][] closeStates() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @Test(dataProvider = "closeStates")
+ public void testCloseReleasesCachedHandles(boolean fenced) throws Exception {
+ ManagedLedgerConfig config = defaultConfig().setRetentionTime(1, TimeUnit.DAYS);
+ ManagedLedger ledger = factory.open("readonly-close-cache", config);
+ Position position = ledger.addEntry(new byte[] {1});
+ ReadOnlyManagedLedgerImpl readOnly = new ReadOnlyManagedLedgerImpl(factory, bkc, factory.getMetaStore(),
+ config, executor, ledger.getName());
+ readOnly.initialize().get(5, TimeUnit.SECONDS);
+ LedgerHandle handle = (LedgerHandle) readOnly.getLedgerHandle(position.getLedgerId()).get(5, TimeUnit.SECONDS);
+ verify(handle, never()).closeAsync();
+ if (fenced) {
+ readOnly.setFenced();
+ assertThatThrownBy(readOnly::close).isInstanceOf(ManagedLedgerFencedException.class);
+ } else {
+ readOnly.close();
+ readOnly.close();
+ }
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> verify(handle).closeAsync());
+ assertThat(readOnly.ledgerCache.size()).isZero();
+ }
+
+ @Test(dataProvider = "closeStates")
+ public void testCloseRejectsSubsequentReadsWithoutOpeningHandles(boolean fenced) throws Exception {
+ ManagedLedgerConfig config = defaultConfig().setRetentionTime(1, TimeUnit.DAYS);
+ ManagedLedger ledger = factory.open("readonly-read-after-close", config);
+ Position position = ledger.addEntry(new byte[] {1});
+ ReadOnlyManagedLedgerImpl readOnly = new ReadOnlyManagedLedgerImpl(factory, bkc, factory.getMetaStore(),
+ config, executor, ledger.getName());
+ readOnly.initialize().get(5, TimeUnit.SECONDS);
+ if (fenced) {
+ readOnly.setFenced();
+ assertThatThrownBy(readOnly::close).isInstanceOf(ManagedLedgerFencedException.class);
+ } else {
+ readOnly.close();
+ }
+ clearInvocations(bkc);
+ CompletableFuture<Void> reading = new CompletableFuture<>();
+ readOnly.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ entry.release();
+ reading.complete(null);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ reading.completeExceptionally(exception);
+ }
+ }, null);
+ assertThatThrownBy(() -> reading.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(ManagedLedgerAlreadyClosedException.class);
+ assertThatThrownBy(() -> readOnly.reopenReadHandle(position.getLedgerId()).get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(ManagedLedgerAlreadyClosedException.class);
+ verify(bkc, never()).newOpenLedgerOp();
+ assertThat(readOnly.ledgerCache.size()).isZero();
+ }
+
+ @Test
+ public void testCloseFailureDoesNotSkipOtherCachedHandles() throws Exception {
+ ManagedLedgerConfig config = defaultConfig().setRetentionTime(1, TimeUnit.DAYS)
+ .setRetentionSizeInMB(-1).setMaxEntriesPerLedger(1);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+ ManagedLedger ledger = factory.open("readonly-close-failure", config);
+ Position first = ledger.addEntry(new byte[] {1});
+ Position second = ledger.addEntry(new byte[] {2});
+ assertThat(first.getLedgerId()).isNotEqualTo(second.getLedgerId());
+ ReadOnlyManagedLedgerImpl readOnly = new ReadOnlyManagedLedgerImpl(factory, bkc, factory.getMetaStore(),
+ config, executor, ledger.getName());
+ readOnly.initialize().get(5, TimeUnit.SECONDS);
+ ReadHandle firstHandle = readOnly.getLedgerHandle(first.getLedgerId()).get(5, TimeUnit.SECONDS);
+ ReadHandle secondHandle = readOnly.getLedgerHandle(second.getLedgerId()).get(5, TimeUnit.SECONDS);
+ doReturn(CompletableFuture.failedFuture(new IllegalStateException("close failed")))
+ .when(firstHandle).closeAsync();
+
+ readOnly.close();
+ readOnly.close();
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(firstHandle).closeAsync();
+ verify(secondHandle).closeAsync();
+ });
+ assertThat(readOnly.ledgerCache.size()).isZero();
+ }
+
+ @Test
+ public void testCloseReleasesHandleWhoseOpenIsPending() throws Exception {
Review Comment:
Nice coverage for a pending BookKeeper open.
One thing I noticed: the future is already in `ledgerCache` before `close()` runs, so this tests cleanup of an existing pending open, but not the race that the second `closingOrClosed` check is meant to handle.
Could we add a small targeted test for a cache insertion racing with close? I'd mainly like to make sure a late insertion can't leave a handle behind.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Fix cached read handle leaks when closing read-only managed ledgers [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26816:
URL: https://github.com/apache/pulsar/pull/26816#issuecomment-6229434311
Thanks @Denovo1998 and @lhotari for the detailed reviews. I've pushed `401ff9c67fd` to address the outstanding feedback:
- added a backward-compatible `ReadOnlyManagedLedger.asyncClose()` and wrapper delegation;
- closed the factory-created ledger from snapshot-segment recovery on every exit path;
- added factory-facing and production-caller lifecycle coverage;
- added deterministic coverage for the concurrent close/open path;
- removed the test's fixed initialization-handle-count assumption; and
- avoided holding the managed-ledger monitor while initiating read-handle cleanup.
The focused managed-ledger and broker tests pass locally, as does `quickCheck`. I also mutation-checked the three new paths: removing the post-open recheck, wrapper delegation, or caller close makes the corresponding test fail.
Could you please take another look when convenient?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Fix cached read handle leaks when closing read-only managed ledgers [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26816:
URL: https://github.com/apache/pulsar/pull/26816#discussion_r4301275078
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java:
##########
@@ -156,6 +160,29 @@ public ReadOnlyCursor createReadOnlyCursor(Position startPosition) {
return new ReadOnlyCursorImpl(bookKeeper, this, startPosition, "read-only-cursor");
}
+ @Override
+ public synchronized void asyncClose(AsyncCallbacks.CloseCallback callback, Object ctx) {
+ closingOrClosed = true;
+ // There is no current write handle, so the parent's write-close callback cannot release this cache.
+ ledgerCache.forEach((ledgerId, readHandle) -> invalidateReadHandle(ledgerId));
+ super.asyncClose(callback, ctx);
Review Comment:
Thanks for pointing this out. Addressed in `401ff9c67fd`: `ReadOnlyManagedLedger` now exposes a backward-compatible default `asyncClose()`, and `ReadOnlyManagedLedgerImplWrapper` delegates it to the underlying implementation. Once the ledger has been acquired, `SnapshotSegmentAbortedTxnProcessorImpl` initiates the close in a `finally` block on every exit path, while logging close failures without masking the recovery result. I also added regression coverage through both the factory API and the production caller.
I kept the factory cursor path separate because a `ReadOnlyCursor` can also be created from a caller-owned `ReadOnlyManagedLedger`, in which case the cursor does not necessarily own the backing ledger. Safely changing `openReadOnlyCursor` would require an ownership-aware wrapper and separate lifecycle coverage. There are currently no in-tree production callers of that factory cursor API; I can address it separately if maintainers would like.
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerHandleCloseTest.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import static org.apache.bookkeeper.mledger.util.ManagedLedgerTestUtil.defaultConfig;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.OpenBuilder;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerAlreadyClosedException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.awaitility.Awaitility;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(timeOut = 30000)
+public class ReadOnlyManagedLedgerHandleCloseTest extends MockedBookKeeperTestCase {
+ private final List<LedgerHandle> openedHandles = new CopyOnWriteArrayList<>();
+ private CompletableFuture<Void> openCompletion;
+
+ @Override
+ protected void setUpTestCase() throws Exception {
+ openedHandles.clear();
+ openCompletion = CompletableFuture.completedFuture(null);
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ bkc = spy(bkc);
+ doAnswer(invocation -> trackOpens((OpenBuilder) invocation.callRealMethod()))
+ .when(bkc).newOpenLedgerOp();
+ factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ }
+
+ @Override
+ protected void cleanUpTestCase() throws Exception {
+ openCompletion.complete(null);
+ // Initialization's temporary handle is covered by a separate fix.
+ for (LedgerHandle handle : openedHandles) {
+ doCallRealMethod().when(handle).closeAsync();
+ handle.closeAsync().get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private OpenBuilder trackOpens(OpenBuilder builder) {
+ OpenBuilder trackedBuilder = spy(builder);
+ doAnswer(invocation -> ((CompletableFuture<ReadHandle>) invocation.callRealMethod()).thenCompose(handle -> {
+ LedgerHandle trackedHandle = spy((LedgerHandle) handle);
+ openedHandles.add(trackedHandle);
+ return openCompletion.thenApply(ignored -> trackedHandle);
+ })).when(trackedBuilder).execute();
+ return trackedBuilder;
+ }
+
+ @DataProvider(name = "closeStates")
+ public Object[][] closeStates() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @Test(dataProvider = "closeStates")
+ public void testCloseReleasesCachedHandles(boolean fenced) throws Exception {
+ ManagedLedgerConfig config = defaultConfig().setRetentionTime(1, TimeUnit.DAYS);
+ ManagedLedger ledger = factory.open("readonly-close-cache", config);
+ Position position = ledger.addEntry(new byte[] {1});
+ ReadOnlyManagedLedgerImpl readOnly = new ReadOnlyManagedLedgerImpl(factory, bkc, factory.getMetaStore(),
+ config, executor, ledger.getName());
+ readOnly.initialize().get(5, TimeUnit.SECONDS);
+ LedgerHandle handle = (LedgerHandle) readOnly.getLedgerHandle(position.getLedgerId()).get(5, TimeUnit.SECONDS);
+ verify(handle, never()).closeAsync();
+ if (fenced) {
+ readOnly.setFenced();
+ assertThatThrownBy(readOnly::close).isInstanceOf(ManagedLedgerFencedException.class);
+ } else {
+ readOnly.close();
+ readOnly.close();
+ }
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> verify(handle).closeAsync());
+ assertThat(readOnly.ledgerCache.size()).isZero();
+ }
+
+ @Test(dataProvider = "closeStates")
+ public void testCloseRejectsSubsequentReadsWithoutOpeningHandles(boolean fenced) throws Exception {
+ ManagedLedgerConfig config = defaultConfig().setRetentionTime(1, TimeUnit.DAYS);
+ ManagedLedger ledger = factory.open("readonly-read-after-close", config);
+ Position position = ledger.addEntry(new byte[] {1});
+ ReadOnlyManagedLedgerImpl readOnly = new ReadOnlyManagedLedgerImpl(factory, bkc, factory.getMetaStore(),
+ config, executor, ledger.getName());
+ readOnly.initialize().get(5, TimeUnit.SECONDS);
+ if (fenced) {
+ readOnly.setFenced();
+ assertThatThrownBy(readOnly::close).isInstanceOf(ManagedLedgerFencedException.class);
+ } else {
+ readOnly.close();
+ }
+ clearInvocations(bkc);
+ CompletableFuture<Void> reading = new CompletableFuture<>();
+ readOnly.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ entry.release();
+ reading.complete(null);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ reading.completeExceptionally(exception);
+ }
+ }, null);
+ assertThatThrownBy(() -> reading.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(ManagedLedgerAlreadyClosedException.class);
+ assertThatThrownBy(() -> readOnly.reopenReadHandle(position.getLedgerId()).get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(ManagedLedgerAlreadyClosedException.class);
+ verify(bkc, never()).newOpenLedgerOp();
+ assertThat(readOnly.ledgerCache.size()).isZero();
+ }
+
+ @Test
+ public void testCloseFailureDoesNotSkipOtherCachedHandles() throws Exception {
+ ManagedLedgerConfig config = defaultConfig().setRetentionTime(1, TimeUnit.DAYS)
+ .setRetentionSizeInMB(-1).setMaxEntriesPerLedger(1);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+ ManagedLedger ledger = factory.open("readonly-close-failure", config);
+ Position first = ledger.addEntry(new byte[] {1});
+ Position second = ledger.addEntry(new byte[] {2});
+ assertThat(first.getLedgerId()).isNotEqualTo(second.getLedgerId());
+ ReadOnlyManagedLedgerImpl readOnly = new ReadOnlyManagedLedgerImpl(factory, bkc, factory.getMetaStore(),
+ config, executor, ledger.getName());
+ readOnly.initialize().get(5, TimeUnit.SECONDS);
+ ReadHandle firstHandle = readOnly.getLedgerHandle(first.getLedgerId()).get(5, TimeUnit.SECONDS);
+ ReadHandle secondHandle = readOnly.getLedgerHandle(second.getLedgerId()).get(5, TimeUnit.SECONDS);
+ doReturn(CompletableFuture.failedFuture(new IllegalStateException("close failed")))
+ .when(firstHandle).closeAsync();
+
+ readOnly.close();
+ readOnly.close();
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(firstHandle).closeAsync();
+ verify(secondHandle).closeAsync();
+ });
+ assertThat(readOnly.ledgerCache.size()).isZero();
+ }
+
+ @Test
+ public void testCloseReleasesHandleWhoseOpenIsPending() throws Exception {
Review Comment:
Thanks. `401ff9c67fd` adds `testCloseRejectsHandleInsertedAfterCloseStarts`, which pauses the BookKeeper open before cache insertion, starts close, waits until the ledger is marked closing, and then allows the open to proceed. It verifies that the returned future fails with `ManagedLedgerAlreadyClosedException`, the late handle is closed, and the cache is empty.
I used two coordinated threads instead of calling `close()` directly inside `newOpenLedgerOp()`, because that mapping function runs while the cache section's non-reentrant write lock is held. Closing inline would attempt to acquire the same lock while draining the cache. I also replaced the fixed handle-count assumption with a recorded baseline. As a mutation check, removing the post-open close-state recheck makes this test fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org