Repository navigation
Conversation
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved ACK handling can report success and advance LAC despite violating fault-domain placement, inline resend callbacks can access cleared or recycled state, and the regression test misses the triggering order.
3 open issues
What changed in this PR
This PR updates the BookKeeper client’s ensemble-change handling to reconcile stale pending adds and invalidate affected acknowledgments before resending writes or advancing completion callbacks.
Changes:
- Reconcile each pending add against the target ensemble.
- Prepare the entire pending queue before resending, then explicitly advance callbacks.
- Add regression tests for stale ensembles, acknowledgment cleanup, late responses, and queued failures.
| File | Description |
|---|---|
bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java |
Adds ensemble reconciliation and callback-ordering regression coverage. |
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java |
Separates ensemble preparation from resending and removes affected successful bookies. |
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java |
Coordinates queue-wide preparation, resends, and callback advancement. |
Review effort: Balanced
| for (int index : resend) { | ||
| addEntrySuccessBookies.remove(previous.get(index)); | ||
| if (!ackSet.removeBookieAndCheck(index)) { | ||
| completed = false; | ||
| } | ||
| } | ||
| return resend; |
| for (int index : indices) { | ||
| sendWriteRequest(ensemble, index); | ||
| } |
| assertTrue(op.completed); | ||
| } | ||
| handle.changingEnsemble = false; | ||
| Set<Integer> changed = new LinkedHashSet<>(Arrays.asList(2, 0, 1)); |
codelipenghui
left a comment
There was a problem hiding this comment.
Reviewed commit 236b9a82a6e36776382a3df7c71e586ce81af0b1.
The per-operation reconciliation and queue-wide ACK preparation are appropriately scoped. Requesting changes for two correctness issues detailed inline:
- P1: Completion can survive removal of an ACK required by the configured fault-domain policy, allowing success/LAC advancement before the replacement acknowledges.
- P2: The multi-index resend loop can continue after an inline callback clears the operation's buffer or recycles its state.
I also left a regression-test comment: the current index order does not exercise the reported premature-callback ordering case.
Validation: These findings are based on source-level execution paths, not a runtime reproduction. The focused Maven attempt selecting PendingAddOpTest and LedgerHandleInlineReadTest stopped in the upstream build-tools module because Surefire found no matching tests there. Neither selected test class ran; no before/after execution was obtained.
| addEntrySuccessBookies.remove(previous.get(index)); | ||
| if (!ackSet.removeBookieAndCheck(index)) { | ||
| completed = false; | ||
| } |
There was a problem hiding this comment.
P1 — Revalidate fault-domain requirements after ACK removal. With fault-domain enforcement enabled, consider ensemble/write quorum 3 and ACK quorum 2: A/B acknowledge in rack 1 and C acknowledges in rack 2 while callbacks are held for an ensemble change. Replacing C removes its ACK, but A/B still satisfy the numeric quorum, so completed remains true. The final sendAddSuccessCallbacks() pass then reports success and advances LAC before the replacement acknowledges, although the remaining successful bookies violate the required rack count.
After all removals, please also clear completed when the enabled placement-policy check rejects addEntrySuccessBookies. Add a two-rack regression that holds the replacement response and asserts that neither the callback nor LAC advances prematurely.
| if (toSend == null || callbackTriggered) { | ||
| return; | ||
| } | ||
|
|
||
|
|
||
| log.debug() | ||
| .ctx(lh.log) | ||
| .attr("entryId", entryId) | ||
| .attr("bookieIndex", bookieIndex) | ||
| .log("Unsetting success"); | ||
|
|
||
|
|
||
| // if we had already heard a success from this array index, need to | ||
| // increment our number of responses that are pending, since we are | ||
| // going to unset this success | ||
| if (!ackSet.removeBookieAndCheck(bookieIndex)) { | ||
| // unset completed if this results in loss of ack quorum | ||
| completed = false; | ||
| for (int index : indices) { | ||
| sendWriteRequest(ensemble, index); |
There was a problem hiding this comment.
P2 — Recheck operation liveness before every resend. BookieClientImpl.addEntry() invokes writeComplete() inline when client-pool lookup fails. If the entry retains ACK quorum after preparation, that callback can deliver its success callback and clear toSend before this loop reaches the next index. For example, with write quorum 4 and ACK quorum 2, retained ACKs at indices 2/3 and retries at 0/1, a synchronous pool-creation failure at 0 can clear the buffer before a cached client for index 1 reaches toSend.retain().
The previous per-index method checked liveness before each send; this loop checks it only once. Please repeat the toSend/callbackTriggered guard inside the loop and add a regression using an initiated operation and a synchronous callback on the first of multiple retries. The operation monitor does not prevent same-thread reentrancy.
| assertTrue(op.completed); | ||
| } | ||
| handle.changingEnsemble = false; | ||
| Set<Integer> changed = new LinkedHashSet<>(Arrays.asList(2, 0, 1)); |
There was a problem hiding this comment.
Regression coverage: Index 2 belongs to both entries' write sets ({1,2} and {2,0}). Processing it first breaks their ACK quorums before a write-set-external index can trigger callbacks, so the callback assertions do not isolate the premature-callback ordering regression.
Please process index 0 before index 2, for example {0,2,1}: index 0 is outside the head entry's write set and exposes callbacks running before the affected ACKs are invalidated. Also cover a resend callback that could drain later pending entries.
codelipenghui
left a comment
There was a problem hiding this comment.
The earlier placement-policy, resend-liveness and test-ordering concerns are addressed in source. Request changes for one new lifecycle regression: moving the pending-write increment before addEntry allows an inline client-close response to recycle the operation while its initial send loop still accesses it, producing an unchecked NullPointerException. Runtime validation could not reach the affected module.
| // The client can report a pool lookup failure inline. | ||
| ++pendingWriteRequests; | ||
| clientCtx.getBookieClient().addEntry(ensemble.get(bookieIndex), | ||
| lh.ledgerId, lh.ledgerKey, entryId, toSend, this, bookieIndex, | ||
| flags, allowFailFast, lh.writeFlags, lh.executor); |
There was a problem hiding this comment.
Protect the initial send loop from inline recycling
Add through a still-open LedgerHandle after its BookKeeper client closes, or race client shutdown with the first initial write. BookieClientImpl then reports ClientClosedException synchronously from addEntry. The error callback completes, but the initial send loop subsequently dereferences a cleared ledger handle and throws NullPointerException. Synchronous addEntry no longer produces the expected BKClientClosedException, and appendAsync can throw instead of returning its exceptionally completed future.
The pre-increment changes the first inline response's pendingWriteRequests transition to 1→0. BookieClientImpl invokes the callback directly when lookup fails, and getRc maps this to ClientClosedException after shutdown. writeComplete decrements the count, then its client-closed branch drains pending adds and delivers callbacks. Because initiate already set hasRun=true, maybeRecycle now recycles the operation and clears lh. Control returns to the unguarded initial loop, whose next condition accesses lh.distributionSchedule. The new liveness guard protects resendEnsembleChange, not initiate. The existing close test explicitly requires normal client-closed error handling for this scenario; it was inspected, not executed.
Keep the counter pre-increment, but prevent initiate from accessing recycled state: check operation liveness before evaluating its loop condition and before each initial send, or defer recycling until initialization exits. Add a deterministic test that uses the actual initiation path and invokes ClientClosedException inline on its first write, asserting one error callback, no unchecked exception and no further sends. Also run BookKeeperCloseTest#testAddLedgerEntry.
Motivation
An add can capture an ensemble before it is queued and miss an ensemble-change resend pass. A later no-op change, or a change affecting only another index, does not retry its missing writes when resend indices are derived only from ledger metadata. The stranded entry blocks subsequent completion callbacks and LAC advancement.
Processing changed indices individually can also invoke completion callbacks for a write-set-external index before ACKs for other changed indices, or later pending entries, have been invalidated.
Changes
Validation
mvn -pl bookkeeper-server -am -Dexec.skip=true -Dtest=PendingAddOpTest,LedgerHandleInlineReadTest -Dsurefire.failIfNoSpecifiedTests=false -DtestRetryCount=0 -Dspotbugs.skip testAll 12 selected tests passed on upstream master (7b66b5d).
Native build commands are skipped for this Java client change. Full-cluster and SpotBugs checks have not been run locally.