You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "merlimat (via GitHub)" <gi...@apache.org> on 2026/09/15 18:27:59 UTC
[PR] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
merlimat opened a new pull request, #26598:
URL: https://github.com/apache/pulsar/pull/26598
### Motivation
Every ledger open in managed-ledger still went through the legacy `BookKeeper.asyncOpenLedger` / `asyncOpenLedgerNoRecovery` overloads, while ledger creation already uses `newCreateLedgerOp()`. The builder API is where BookKeeper adds per-ledger options: 4.18.1, now on master, ships `OpenBuilder.withKeepUpdateMetadata` (apache/bookkeeper#4834) and `withOrderingKey` on the create and open builders (apache/bookkeeper#4881). Moving the remaining opens to `newOpenLedgerOp()` is the prerequisite for using those options. This PR only switches the API and does not change behavior.
### Modifications
- `ManagedLedgerImpl` (init-time open of the last ledger, concurrent-modification recheck) and `ManagedCursorImpl` (cursor-ledger recovery): `newOpenLedgerOp().withRecovery(true).withKeepUpdateMetadata(true)`, which maps onto the same `LedgerOpenOp.initiateWithKeepUpdateMetadata()` path as the previous `asyncOpenLedger(..., keepUpdateMetadata=true)` call.
- `ShadowManagedLedgerImpl` (source ledger opens) and `ManagedLedgerFactoryImpl` (offline topic stats cursor ledger): `newOpenLedgerOp().withRecovery(false)`.
- The existing `OpenCallback` bodies are unchanged; the returned future is adapted with `BKException.getExceptionCode`, and the handle is cast to `LedgerHandle` since the real client returns a `ReadOnlyLedgerHandle`.
- `testmocks`: `PulsarMockBookKeeper.newOpenLedgerOp()` returned a plain `ReadHandle`, which cannot be cast to `LedgerHandle`. `PulsarMockReadHandle` is now a read-only `LedgerHandle` sharing the writer's entries, like `ReadOnlyLedgerHandle`: close is a no-op, reads still go through the read interceptor, it honors the mock's empty-ledger countdown, and the legacy `asyncReadEntries` is shared with `PulsarMockLedgerHandle`. The mock's open builder completes on the mock executor, as the legacy mock path and the real client do.
- `ManagedCursorTest` and `ManagedLedgerFactoryShutdownTest` intercept `newOpenLedgerOp()` instead of the legacy overloads.
Broker-side users of `asyncOpenLedger` (schema storage, bucket snapshots, compaction) are unchanged; they do not run on a managed-ledger thread.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change is already covered by existing tests, such as `ManagedCursorTest`, `ManagedLedgerTest`, `ManagedLedgerBkTest`, `ManagedLedgerFactoryChangeLedgerPathTest`, `ShadowManagedLedgerImplTest`, `ReadOnlyManagedLedgerImplTest`, `ManagedLedgerFactoryShutdownTest`, `BrokerServiceTest` and `BrokerEntryCacheTest`.
--
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] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #26598:
URL: https://github.com/apache/pulsar/pull/26598
--
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] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26598:
URL: https://github.com/apache/pulsar/pull/26598#discussion_r4020126295
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java:
##########
@@ -18,40 +18,48 @@
*/
package org.apache.bookkeeper.client;
+import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import lombok.CustomLog;
+import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
+import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.api.LedgerMetadata;
-import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.client.impl.LedgerEntriesImpl;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Versioned;
/**
- * Mock implementation of ReadHandle.
+ * Mock read-only view of a ledger, the counterpart of {@link ReadOnlyLedgerHandle}.
+ *
+ * <p>It is a {@link LedgerHandle} sharing the entries of the {@link PulsarMockLedgerHandle} that wrote them, so that
+ * what {@link PulsarMockBookKeeper#newOpenLedgerOp()} returns can be cast to {@link LedgerHandle} like the handle
+ * returned by the real client. Closing it never affects the ledger.
*/
@CustomLog
-class PulsarMockReadHandle implements ReadHandle {
+class PulsarMockReadHandle extends LedgerHandle {
Review Comment:
This is not a real issue since tests pass.
--
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] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26598:
URL: https://github.com/apache/pulsar/pull/26598#discussion_r4020123348
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java:
##########
@@ -323,7 +324,7 @@ public OpenBuilder newOpenLedgerOp() {
return new OpenBuilderBase() {
@Override
public CompletableFuture<ReadHandle> execute() {
- return getProgrammedFailure().thenCompose(
+ return getProgrammedFailure().thenComposeAsync(
Review Comment:
Since current tests pass, this doesn't seem to be a real issue.
--
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] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26598:
URL: https://github.com/apache/pulsar/pull/26598#discussion_r4020050825
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java:
##########
@@ -323,7 +324,7 @@ public OpenBuilder newOpenLedgerOp() {
return new OpenBuilderBase() {
@Override
public CompletableFuture<ReadHandle> execute() {
- return getProgrammedFailure().thenCompose(
+ return getProgrammedFailure().thenComposeAsync(
Review Comment:
**[QUALITY] PulsarMockBookKeeper.newOpenLedgerOp() no longer dispatches onto the mock executor for a programmed failure (failNow/failAfter), contradicting the PR's own description that it "completes on the mock executor, as the legacy mock path ... do[es]"**
`getProgrammedFailure().thenComposeAsync(fn, executor)`: when `failNow`/`failAfter` have already completed the programmed-failure future *exceptionally* (which is how they work — the promise is completed at programming time, before any ledger is opened), `thenComposeAsync` short-circuits and completes the dependent stage inline, without ever submitting `fn` to `executor`. The old mock open path ended with `whenCompleteAsync(..., executor)`, which — being an `Async`-suffixed method — always submits to the given executor even when the source is already complete.
So for a programmed failure specifically, the open's completion callback now runs inline on whichever thread completes the programmed-failure promise (e.g. the test thread, or whatever thread is inside a `synchronized` managed-ledger method that triggered the open), instead of on the mock's own executor thread the way it did before and the way normally-completing opens still do. Tests that arm `bkc.failNow(rc)`/`bkc.failAfter(...)` before a converted open and rely on the old async hand-off for thread-interleaving may see different ordering/reentrancy.
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java:
##########
@@ -18,40 +18,48 @@
*/
package org.apache.bookkeeper.client;
+import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import lombok.CustomLog;
+import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
+import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.api.LedgerMetadata;
-import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.client.impl.LedgerEntriesImpl;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Versioned;
/**
- * Mock implementation of ReadHandle.
+ * Mock read-only view of a ledger, the counterpart of {@link ReadOnlyLedgerHandle}.
+ *
+ * <p>It is a {@link LedgerHandle} sharing the entries of the {@link PulsarMockLedgerHandle} that wrote them, so that
+ * what {@link PulsarMockBookKeeper#newOpenLedgerOp()} returns can be cast to {@link LedgerHandle} like the handle
+ * returned by the real client. Closing it never affects the ledger.
*/
@CustomLog
-class PulsarMockReadHandle implements ReadHandle {
+class PulsarMockReadHandle extends LedgerHandle {
Review Comment:
**[QUALITY] PulsarMockReadHandle now extends LedgerHandle without overriding the write path, so it no longer enforces the read-only invariant the real client's ReadOnlyLedgerHandle enforces**
Before this PR, `PulsarMockReadHandle` implemented only the narrow `ReadHandle` interface, so calling a write method on it was a compile error. Now it extends `LedgerHandle` (needed so the mock's `newOpenLedgerOp()` result can be cast to `LedgerHandle` like the real client's `ReadOnlyLedgerHandle`), but — unlike `ReadOnlyLedgerHandle`, which overrides `addEntry`/`asyncAddEntry` to reject writes with `BKException.Code.IllegalOpException` — it doesn't override any write method.
For an open, unclosed mock ledger, calling `addEntry`/`asyncAddEntry` on a handle obtained from `newOpenLedgerOp()` now falls through to `LedgerHandle`'s real write path (`doAsyncAddEntry` → `PendingAddOp` → `clientCtx.getBookieClient().addEntry(...)`), which throws a synchronous `NullPointerException` because the mock's client context has a null `BookieClient`/`LedgerManager` — instead of the clean `IllegalOpException` callback the real client would give. I didn't find any current managed-ledger/broker call site that writes through one of these read-only-opened handles, so this isn't a live bug, but it does weaken the mock's fidelity: a future write-on-a-read-only-handle mistake (production code or a new test) would now NPE unpredictably instead of getting the same clean rejection the real client gives.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -508,7 +508,16 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
log.debug().attr("ledgerId", id).log("Opening ledger");
mbean.startDataLedgerOpenOp();
- bookKeeper.asyncOpenLedger(id, digestType, config.getPassword(), opencb, null, true);
+ bookKeeper.newOpenLedgerOp()
+ .withRecovery(true)
+ .withLedgerId(id)
+ .withDigestType(config.getDigestType())
+ .withPassword(config.getPassword())
+ .withKeepUpdateMetadata(true)
+ .withLoggerContext(log)
+ .execute()
+ .whenComplete((rh, ex) ->
+ opencb.openComplete(BKException.getExceptionCode(ex), (LedgerHandle) rh, null));
Review Comment:
**[BUG] Every converted newOpenLedgerOp()...execute().whenComplete(...) call discards the returned stage, so an exception thrown inside the completion callback now disappears silently instead of being logged**
None of the five converted call sites ([`ManagedLedgerImpl.java:519`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L519) and [`ManagedLedgerImpl.java:1931`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L1931), [`ManagedCursorImpl.java:701`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L701), [`ManagedLedgerFactoryImpl.java:1494`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java#L1494), [`ShadowManagedLedgerImpl.java:172`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/man
aged-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java#L172) and [`ShadowManagedLedgerImpl.java:365`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java#L365)) keep the `CompletableFuture` that `.whenComplete()` returns. Per the `CompletableFuture` contract, if the action itself throws, that exception is captured into the *returned* stage rather than surfacing on the completing thread — and with nothing chained onto the returned stage, the exception is never logged or otherwise observed.
Before this PR, a successful open's callback ran through `LedgerHandle.executeOrdered(...)`, i.e. BookKeeper's `OrderedExecutor` — its `SingleThreadExecutor.safeRunTask` explicitly logs any exception that escapes a submitted task. That diagnostic is gone after the refactor. At [`ManagedCursorImpl.java:692-705`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L692-L705) the surrounding `try { ... } catch (Throwable t) { ... }` doesn't help either: it only wraps the synchronous chain-building up to where `.whenComplete()` returns a new stage — the lambda itself runs later, outside that try block, on whatever thread completes the open.
Concrete trigger: `opencb`'s body at [`ManagedLedgerImpl.java:464`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L464) does `executor.execute(() -> {...})`. If the managed-ledger's own executor has already been shut down by the time the open completes (e.g. a slow BK ledger open racing a ledger/topic close), that `executor.execute(...)` throws `RejectedExecutionException` synchronously inside the `whenComplete` action — and it now vanishes instead of being logged the way the old executor-backed callback would have logged it. The same discard pattern applies at the other four call sites for whatever their callback bodies (`ledgerClosed`, cursor `initialize`/`recoveredCursor`, etc.) might throw.
A simple fix is a try/catch that logs inside each lambda. Chaining `.exceptionally(...)` onto the returned stage is not equivalent: that stage relays the *open's* own failure too, so it would log every ordinary failed open a second time next to the rc handling the callback already does.
--
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] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on code in PR #26598:
URL: https://github.com/apache/pulsar/pull/26598#discussion_r4020184782
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -508,7 +508,16 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
log.debug().attr("ledgerId", id).log("Opening ledger");
mbean.startDataLedgerOpenOp();
- bookKeeper.asyncOpenLedger(id, digestType, config.getPassword(), opencb, null, true);
+ bookKeeper.newOpenLedgerOp()
+ .withRecovery(true)
+ .withLedgerId(id)
+ .withDigestType(config.getDigestType())
+ .withPassword(config.getPassword())
+ .withKeepUpdateMetadata(true)
+ .withLoggerContext(log)
+ .execute()
+ .whenComplete((rh, ex) ->
+ opencb.openComplete(BKException.getExceptionCode(ex), (LedgerHandle) rh, null));
Review Comment:
Good catch, thanks. Fixed in the follow-up commit: the six sites now go through a small `ManagedLedgerImpl.completeOpenCallback(log, ledgerId, callback, handle, ex)` bridge that invokes the legacy callback inside a try/catch and logs whatever it throws (so a `RejectedExecutionException` from the ML executor is logged again, as it was under BK's `safeRunTask`). I went with the shared helper rather than a try/catch per lambda, and kept `.exceptionally(...)` out for the reason you gave: it would double-log every ordinary failed open.
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java:
##########
@@ -323,7 +324,7 @@ public OpenBuilder newOpenLedgerOp() {
return new OpenBuilderBase() {
@Override
public CompletableFuture<ReadHandle> execute() {
- return getProgrammedFailure().thenCompose(
+ return getProgrammedFailure().thenComposeAsync(
Review Comment:
Agreed it's not blocking since the tests pass. I still tightened it in the follow-up commit: `newOpenLedgerOp().execute()` now completes a fresh future from `getProgrammedFailure().whenCompleteAsync(..., executor)`, so a programmed failure also hands off to the mock executor instead of completing inline on the caller's thread, matching the legacy `asyncOpenLedger` mock path.
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java:
##########
@@ -18,40 +18,48 @@
*/
package org.apache.bookkeeper.client;
+import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import lombok.CustomLog;
+import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
+import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.api.LedgerMetadata;
-import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.client.impl.LedgerEntriesImpl;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Versioned;
/**
- * Mock implementation of ReadHandle.
+ * Mock read-only view of a ledger, the counterpart of {@link ReadOnlyLedgerHandle}.
+ *
+ * <p>It is a {@link LedgerHandle} sharing the entries of the {@link PulsarMockLedgerHandle} that wrote them, so that
+ * what {@link PulsarMockBookKeeper#newOpenLedgerOp()} returns can be cast to {@link LedgerHandle} like the handle
+ * returned by the real client. Closing it never affects the ledger.
*/
@CustomLog
-class PulsarMockReadHandle implements ReadHandle {
+class PulsarMockReadHandle extends LedgerHandle {
Review Comment:
Agreed it's not blocking. Since the point of the read view is to behave like `ReadOnlyLedgerHandle`, the follow-up commit makes `PulsarMockReadHandle` override `addEntry`/`asyncAddEntry` (including the `ByteBuf` variant) to fail with `IllegalOpException`, so a stray write on a handle from `newOpenLedgerOp()` gets the same clean rejection as with the real client instead of an NPE.
--
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] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26598:
URL: https://github.com/apache/pulsar/pull/26598#discussion_r4020056868
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java:
##########
@@ -18,40 +18,48 @@
*/
package org.apache.bookkeeper.client;
+import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import lombok.CustomLog;
+import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
+import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.api.LedgerMetadata;
-import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.client.impl.LedgerEntriesImpl;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Versioned;
/**
- * Mock implementation of ReadHandle.
+ * Mock read-only view of a ledger, the counterpart of {@link ReadOnlyLedgerHandle}.
+ *
+ * <p>It is a {@link LedgerHandle} sharing the entries of the {@link PulsarMockLedgerHandle} that wrote them, so that
+ * what {@link PulsarMockBookKeeper#newOpenLedgerOp()} returns can be cast to {@link LedgerHandle} like the handle
+ * returned by the real client. Closing it never affects the ledger.
*/
@CustomLog
-class PulsarMockReadHandle implements ReadHandle {
+class PulsarMockReadHandle extends LedgerHandle {
Review Comment:
**[QUALITY] PulsarMockReadHandle now extends LedgerHandle without overriding the write path, so it no longer enforces the read-only invariant the real client's ReadOnlyLedgerHandle enforces**
Before this PR, `PulsarMockReadHandle` implemented only the narrow `ReadHandle` interface, so calling a write method on it was a compile error. Now it extends `LedgerHandle` (needed so the mock's `newOpenLedgerOp()` result can be cast to `LedgerHandle` like the real client's `ReadOnlyLedgerHandle`), but — unlike `ReadOnlyLedgerHandle`, which overrides `addEntry`/`asyncAddEntry` to reject writes with `BKException.Code.IllegalOpException` — it doesn't override any write method.
For an open, unclosed mock ledger, calling `addEntry`/`asyncAddEntry` on a handle obtained from `newOpenLedgerOp()` now falls through to `LedgerHandle`'s real write path (`doAsyncAddEntry` → `PendingAddOp` → `clientCtx.getBookieClient().addEntry(...)`), which throws a synchronous `NullPointerException` because the mock's client context has a null `BookieClient`/`LedgerManager` — instead of the clean `IllegalOpException` callback the real client would give. I didn't find any current managed-ledger/broker call site that writes through one of these read-only-opened handles, so this isn't a live bug, but it does weaken the mock's fidelity: a future write-on-a-read-only-handle mistake (production code or a new test) would now NPE unpredictably instead of getting the same clean rejection the real client gives.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -508,7 +508,16 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
log.debug().attr("ledgerId", id).log("Opening ledger");
mbean.startDataLedgerOpenOp();
- bookKeeper.asyncOpenLedger(id, digestType, config.getPassword(), opencb, null, true);
+ bookKeeper.newOpenLedgerOp()
+ .withRecovery(true)
+ .withLedgerId(id)
+ .withDigestType(config.getDigestType())
+ .withPassword(config.getPassword())
+ .withKeepUpdateMetadata(true)
+ .withLoggerContext(log)
+ .execute()
+ .whenComplete((rh, ex) ->
+ opencb.openComplete(BKException.getExceptionCode(ex), (LedgerHandle) rh, null));
Review Comment:
**[BUG] Every converted newOpenLedgerOp()...execute().whenComplete(...) call discards the returned stage, so an exception thrown inside the completion callback now disappears silently instead of being logged**
None of the five converted call sites ([`ManagedLedgerImpl.java:519`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L519) and [`ManagedLedgerImpl.java:1931`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L1931), [`ManagedCursorImpl.java:701`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L701), [`ManagedLedgerFactoryImpl.java:1494`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java#L1494), [`ShadowManagedLedgerImpl.java:172`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/man
aged-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java#L172) and [`ShadowManagedLedgerImpl.java:365`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java#L365)) keep the `CompletableFuture` that `.whenComplete()` returns. Per the `CompletableFuture` contract, if the action itself throws, that exception is captured into the *returned* stage rather than surfacing on the completing thread — and with nothing chained onto the returned stage, the exception is never logged or otherwise observed.
Before this PR, a successful open's callback ran through `LedgerHandle.executeOrdered(...)`, i.e. BookKeeper's `OrderedExecutor` — its `SingleThreadExecutor.safeRunTask` explicitly logs any exception that escapes a submitted task. That diagnostic is gone after the refactor. At [`ManagedCursorImpl.java:692-705`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L692-L705) the surrounding `try { ... } catch (Throwable t) { ... }` doesn't help either: it only wraps the synchronous chain-building up to where `.whenComplete()` returns a new stage — the lambda itself runs later, outside that try block, on whatever thread completes the open.
Concrete trigger: `opencb`'s body at [`ManagedLedgerImpl.java:464`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L464) does `executor.execute(() -> {...})`. If the managed-ledger's own executor has already been shut down by the time the open completes (e.g. a slow BK ledger open racing a ledger/topic close), that `executor.execute(...)` throws `RejectedExecutionException` synchronously inside the `whenComplete` action — and it now vanishes instead of being logged the way the old executor-backed callback would have logged it. The same discard pattern applies at the other four call sites for whatever their callback bodies (`ledgerClosed`, cursor `initialize`/`recoveredCursor`, etc.) might throw.
A simple fix is to chain `.exceptionally(t -> { log.error()...; return null; })` (or fold a try/catch into the lambda) after each `.whenComplete(...)`.
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java:
##########
@@ -323,7 +324,7 @@ public OpenBuilder newOpenLedgerOp() {
return new OpenBuilderBase() {
@Override
public CompletableFuture<ReadHandle> execute() {
- return getProgrammedFailure().thenCompose(
+ return getProgrammedFailure().thenComposeAsync(
Review Comment:
**[QUALITY] PulsarMockBookKeeper.newOpenLedgerOp() no longer dispatches onto the mock executor for a programmed failure (failNow/failAfter), contradicting the PR's own description that it "completes on the mock executor, as the legacy mock path ... do[es]"**
`getProgrammedFailure().thenComposeAsync(fn, executor)`: when `failNow`/`failAfter` have already completed the programmed-failure future *exceptionally* (which is how they work — the promise is completed at programming time, before any ledger is opened), `thenComposeAsync` short-circuits and completes the dependent stage inline, without ever submitting `fn` to `executor`. The old mock open path ended with `whenCompleteAsync(..., executor)`, which — being an `Async`-suffixed method — always submits to the given executor even when the source is already complete.
So for a programmed failure specifically, the open's completion callback now runs inline on whichever thread completes the programmed-failure promise (e.g. the test thread, or whatever thread is inside a `synchronized` managed-ledger method that triggered the open), instead of on the mock's own executor thread the way it did before and the way normally-completing opens still do. Tests that arm `bkc.failNow(rc)`/`bkc.failAfter(...)` before a converted open and rely on the old async hand-off for thread-interleaving may see different ordering/reentrancy.
--
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] [improve][ml] Open ledgers through the BookKeeper builder API [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26598:
URL: https://github.com/apache/pulsar/pull/26598#discussion_r4020056854
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -508,7 +508,16 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
log.debug().attr("ledgerId", id).log("Opening ledger");
mbean.startDataLedgerOpenOp();
- bookKeeper.asyncOpenLedger(id, digestType, config.getPassword(), opencb, null, true);
+ bookKeeper.newOpenLedgerOp()
+ .withRecovery(true)
+ .withLedgerId(id)
+ .withDigestType(config.getDigestType())
+ .withPassword(config.getPassword())
+ .withKeepUpdateMetadata(true)
+ .withLoggerContext(log)
+ .execute()
+ .whenComplete((rh, ex) ->
+ opencb.openComplete(BKException.getExceptionCode(ex), (LedgerHandle) rh, null));
Review Comment:
**[BUG] Every converted newOpenLedgerOp()...execute().whenComplete(...) call discards the returned stage, so an exception thrown inside the completion callback now disappears silently instead of being logged**
None of the five converted call sites ([`ManagedLedgerImpl.java:519`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L519) and [`ManagedLedgerImpl.java:1931`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L1931), [`ManagedCursorImpl.java:701`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L701), [`ManagedLedgerFactoryImpl.java:1494`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java#L1494), [`ShadowManagedLedgerImpl.java:172`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/man
aged-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java#L172) and [`ShadowManagedLedgerImpl.java:365`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java#L365)) keep the `CompletableFuture` that `.whenComplete()` returns. Per the `CompletableFuture` contract, if the action itself throws, that exception is captured into the *returned* stage rather than surfacing on the completing thread — and with nothing chained onto the returned stage, the exception is never logged or otherwise observed.
Before this PR, a successful open's callback ran through `LedgerHandle.executeOrdered(...)`, i.e. BookKeeper's `OrderedExecutor` — its `SingleThreadExecutor.safeRunTask` explicitly logs any exception that escapes a submitted task. That diagnostic is gone after the refactor. At [`ManagedCursorImpl.java:692-705`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L692-L705) the surrounding `try { ... } catch (Throwable t) { ... }` doesn't help either: it only wraps the synchronous chain-building up to where `.whenComplete()` returns a new stage — the lambda itself runs later, outside that try block, on whatever thread completes the open.
Concrete trigger: `opencb`'s body at [`ManagedLedgerImpl.java:464`](https://github.com/apache/pulsar/blob/39fd7968bfc17be183cd2a1a0eeff3dce879243b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L464) does `executor.execute(() -> {...})`. If the managed-ledger's own executor has already been shut down by the time the open completes (e.g. a slow BK ledger open racing a ledger/topic close), that `executor.execute(...)` throws `RejectedExecutionException` synchronously inside the `whenComplete` action — and it now vanishes instead of being logged the way the old executor-backed callback would have logged it. The same discard pattern applies at the other four call sites for whatever their callback bodies (`ledgerClosed`, cursor `initialize`/`recoveredCursor`, etc.) might throw.
A simple fix is to chain `.exceptionally(t -> { log.error()...; return null; })` (or fold a try/catch into the lambda) after each `.whenComplete(...)`.
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java:
##########
@@ -323,7 +324,7 @@ public OpenBuilder newOpenLedgerOp() {
return new OpenBuilderBase() {
@Override
public CompletableFuture<ReadHandle> execute() {
- return getProgrammedFailure().thenCompose(
+ return getProgrammedFailure().thenComposeAsync(
Review Comment:
**[QUALITY] PulsarMockBookKeeper.newOpenLedgerOp() no longer dispatches onto the mock executor for a programmed failure (failNow/failAfter), contradicting the PR's own description that it "completes on the mock executor, as the legacy mock path ... do[es]"**
`getProgrammedFailure().thenComposeAsync(fn, executor)`: when `failNow`/`failAfter` have already completed the programmed-failure future *exceptionally* (which is how they work — the promise is completed at programming time, before any ledger is opened), `thenComposeAsync` short-circuits and completes the dependent stage inline, without ever submitting `fn` to `executor`. The old mock open path ended with `whenCompleteAsync(..., executor)`, which — being an `Async`-suffixed method — always submits to the given executor even when the source is already complete.
So for a programmed failure specifically, the open's completion callback now runs inline on whichever thread completes the programmed-failure promise (e.g. the test thread, or whatever thread is inside a `synchronized` managed-ledger method that triggered the open), instead of on the mock's own executor thread the way it did before and the way normally-completing opens still do. Tests that arm `bkc.failNow(rc)`/`bkc.failAfter(...)` before a converted open and rely on the old async hand-off for thread-interleaving may see different ordering/reentrancy.
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java:
##########
@@ -18,40 +18,48 @@
*/
package org.apache.bookkeeper.client;
+import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import lombok.CustomLog;
+import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
+import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.api.LedgerMetadata;
-import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.client.impl.LedgerEntriesImpl;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Versioned;
/**
- * Mock implementation of ReadHandle.
+ * Mock read-only view of a ledger, the counterpart of {@link ReadOnlyLedgerHandle}.
+ *
+ * <p>It is a {@link LedgerHandle} sharing the entries of the {@link PulsarMockLedgerHandle} that wrote them, so that
+ * what {@link PulsarMockBookKeeper#newOpenLedgerOp()} returns can be cast to {@link LedgerHandle} like the handle
+ * returned by the real client. Closing it never affects the ledger.
*/
@CustomLog
-class PulsarMockReadHandle implements ReadHandle {
+class PulsarMockReadHandle extends LedgerHandle {
Review Comment:
**[QUALITY] PulsarMockReadHandle now extends LedgerHandle without overriding the write path, so it no longer enforces the read-only invariant the real client's ReadOnlyLedgerHandle enforces**
Before this PR, `PulsarMockReadHandle` implemented only the narrow `ReadHandle` interface, so calling a write method on it was a compile error. Now it extends `LedgerHandle` (needed so the mock's `newOpenLedgerOp()` result can be cast to `LedgerHandle` like the real client's `ReadOnlyLedgerHandle`), but — unlike `ReadOnlyLedgerHandle`, which overrides `addEntry`/`asyncAddEntry` to reject writes with `BKException.Code.IllegalOpException` — it doesn't override any write method.
For an open, unclosed mock ledger, calling `addEntry`/`asyncAddEntry` on a handle obtained from `newOpenLedgerOp()` now falls through to `LedgerHandle`'s real write path (`doAsyncAddEntry` → `PendingAddOp` → `clientCtx.getBookieClient().addEntry(...)`), which throws a synchronous `NullPointerException` because the mock's client context has a null `BookieClient`/`LedgerManager` — instead of the clean `IllegalOpException` callback the real client would give. I didn't find any current managed-ledger/broker call site that writes through one of these read-only-opened handles, so this isn't a live bug, but it does weaken the mock's fidelity: a future write-on-a-read-only-handle mistake (production code or a new test) would now NPE unpredictably instead of getting the same clean rejection the real client gives.
--
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