This is an automated email from the ASF dual-hosted git repository. merlimat pushed a commit to branch branch-4.17 in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
commit 3ab22084f2edb19d1d21a13061f00cfaed352e54 Author: Enrico Olivelli <[email protected]> AuthorDate: Thu Apr 30 00:23:34 2026 +0200 Fix NPE in PendingAddOp.maybeTimeout() when clientCtx is null after recycling (#4760) * Fix NPE in PendingAddOp.maybeTimeout() when clientCtx is null after recycling monitorPendingAddOps() iterates the pendingAddOps queue and calls maybeTimeout() on each op. Concurrently, sendAddSuccessCallbacks() can remove a completed op from the queue and, via submitCallback(), lead to recyclePendAddOpObject() which sets clientCtx = null. If the scheduler thread still holds an iterator reference to that op and then calls maybeTimeout(), the dereference of clientCtx causes an NPE. The race is triggered in practice when addEntryQuorumTimeoutNanos > 0 (i.e. the quorum-timeout monitor is enabled). It became visible with Netty 4.1.130, which changed Recycler thread-scheduling behavior and made the narrow window between the iterator snapshot and the null assignment observable. Fix: - Make clientCtx volatile so that the null write in the synchronized recyclePendAddOpObject() is immediately visible to the unsynchronized maybeTimeout() reader. - Guard the top of maybeTimeout() with an explicit null check: if clientCtx is null the op has already completed and recycled, so there is nothing to time out and the method returns false. Closes: https://github.com/apache/bookkeeper/issues/4759 * Address review: snapshot clientCtx and guard timeoutQuorumWait against recycle Per @merlimat's review on #4760, the original null-check in maybeTimeout() still races: clientCtx may be nulled between the guard and the subsequent getConf() dereference. Volatile only addresses visibility, not atomicity across two reads. - Snapshot clientCtx into a local in maybeTimeout(); the local cannot be mutated by another thread, so the dereference is race-free. - Drop the volatile modifier on clientCtx (no longer load-bearing once the read is done once into a local). - Extend timeoutQuorumWait()'s early-return to also short-circuit when lh / clientCtx are null. recyclePendAddOpObject() resets `completed` to false, so the prior `if (completed)` guard does not cover the already-recycled case; without this check timeoutQuorumWait() NPEs on lh.getLedgerMetadata() (and worse, could invoke handleUnrecoverableErrorDuringAdd on a stale handle). - Add testTimeoutQuorumWaitIsNoOpWhenAlreadyRecycled covering the new guard. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> --------- Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]> --- .../org/apache/bookkeeper/client/PendingAddOp.java | 17 +++- .../apache/bookkeeper/client/PendingAddOpTest.java | 110 +++++++++++++++++++++ 2 files changed, 125 insertions(+), 2 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java index 51f559a86c..1df221e2ea 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java @@ -154,7 +154,16 @@ class PendingAddOp implements WriteCallback { } boolean maybeTimeout() { - if (MathUtils.elapsedNanos(requestTimeNanos) >= clientCtx.getConf().addEntryQuorumTimeoutNanos) { + // Snapshot clientCtx into a local: recyclePendAddOpObject() may run on another thread + // and null the field while monitorPendingAddOps() still holds an iterator reference + // to this op. A single read prevents the field from going null between the guard and + // the getConf() dereference below. + ClientContext ctx = clientCtx; + if (ctx == null) { + // Already recycled — the add-entry completed before the timeout monitor fired. + return false; + } + if (MathUtils.elapsedNanos(requestTimeNanos) >= ctx.getConf().addEntryQuorumTimeoutNanos) { timeoutQuorumWait(); return true; } @@ -162,7 +171,11 @@ class PendingAddOp implements WriteCallback { } synchronized void timeoutQuorumWait() { - if (completed) { + // lh / clientCtx are nulled by recyclePendAddOpObject() under this same monitor. + // Once we hold the lock, a recycle has either not started or fully completed; if any + // of these are null, the op has been recycled and there is nothing to time out. + // (The completed flag alone is insufficient: recycle resets it to false.) + if (completed || lh == null || clientCtx == null) { return; } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java index 5fb318c51f..968bbaab56 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java @@ -20,17 +20,25 @@ package org.apache.bookkeeper.client; import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import java.util.concurrent.atomic.AtomicInteger; import org.apache.bookkeeper.client.BKException.Code; +import org.apache.bookkeeper.client.api.LedgerMetadata; import org.apache.bookkeeper.client.api.WriteFlag; +import org.apache.bookkeeper.common.util.MathUtils; import org.apache.bookkeeper.common.util.OrderedExecutor; +import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.proto.BookieClient; import org.apache.bookkeeper.stats.NullStatsLogger; import org.junit.Before; @@ -87,4 +95,106 @@ public class PendingAddOpTest { assertNull(op.lh); } + /** + * Verify that {@link PendingAddOp#maybeTimeout()} returns {@code false} without throwing + * {@link NullPointerException} when {@code clientCtx} is {@code null}. + * + * <p>This is the exact scenario that caused the production NPE reported in + * apache/bookkeeper#4759: {@code monitorPendingAddOps()} holds an iterator reference to + * an op while {@code recyclePendAddOpObject()} runs concurrently on another thread and + * clears {@code clientCtx} to {@code null}. After recycling the op has already completed, + * so the correct behaviour is to skip the timeout check (return {@code false}). + */ + @Test + public void testMaybeTimeoutReturnsFalseWhenClientCtxIsNull() { + PendingAddOp op = PendingAddOp.create( + lh, mockClientContext, lh.getCurrentEnsemble(), + payload, WriteFlag.NONE, + (rc, handle, entryId, qwcLatency, ctx) -> {}, null); + + // Simulate the race: recyclePendAddOpObject() cleared clientCtx on the writer + // thread while monitorPendingAddOps() is still iterating the pendingAddOps queue + // on the scheduler thread. + op.clientCtx = null; + + // Before the fix this threw NullPointerException; after the fix it must return false. + assertFalse(op.maybeTimeout()); + } + + /** + * Verify that {@link PendingAddOp#maybeTimeout()} returns {@code false} when + * {@code clientCtx} is non-null and the quorum timeout has not yet elapsed. + */ + @Test + public void testMaybeTimeoutReturnsFalseWhenWithinQuorumTimeout() { + // Configure a 1-hour quorum timeout so the op will not have expired. + ClientConfiguration conf = new ClientConfiguration(); + conf.setAddEntryQuorumTimeout(3600); + when(mockClientContext.getConf()).thenReturn(ClientInternalConf.fromConfig(conf)); + + PendingAddOp op = PendingAddOp.create( + lh, mockClientContext, lh.getCurrentEnsemble(), + payload, WriteFlag.NONE, + (rc, handle, entryId, qwcLatency, ctx) -> {}, null); + // Stamp the request as starting right now so elapsed time is effectively 0. + op.requestTimeNanos = MathUtils.nowInNano(); + + assertFalse(op.maybeTimeout()); + } + + /** + * Verify that {@link PendingAddOp#maybeTimeout()} returns {@code true} and triggers + * {@link PendingAddOp#timeoutQuorumWait()} when the quorum timeout has already elapsed. + */ + @Test + public void testMaybeTimeoutReturnsTrueWhenQuorumTimeoutExpired() { + // Configure a 1-second quorum timeout. + ClientConfiguration conf = new ClientConfiguration(); + conf.setAddEntryQuorumTimeout(1); + when(mockClientContext.getConf()).thenReturn(ClientInternalConf.fromConfig(conf)); + + LedgerMetadata meta = mock(LedgerMetadata.class); + when(lh.getLedgerMetadata()).thenReturn(meta); + // addEntrySuccessBookies starts empty (size 0 < ackQuorumSize 3), + // so the fault-domain stats branch is skipped and no extra mocking is needed. + when(meta.getAckQuorumSize()).thenReturn(3); + + PendingAddOp op = PendingAddOp.create( + lh, mockClientContext, lh.getCurrentEnsemble(), + payload, WriteFlag.NONE, + (rc, handle, entryId, qwcLatency, ctx) -> {}, null); + // Set the request start time to 0 so that elapsed nanos is enormous (>> 1 second). + op.requestTimeNanos = 0L; + + assertTrue(op.maybeTimeout()); + } + + /** + * Verify that {@link PendingAddOp#timeoutQuorumWait()} is a no-op when the op has already + * been recycled (i.e. {@code lh} and {@code clientCtx} are {@code null}). + * + * <p>This covers the second concern raised in the review of #4760: even after + * {@link PendingAddOp#maybeTimeout()} captures a non-null {@code clientCtx} snapshot, + * {@code recyclePendAddOpObject()} may complete before {@code timeoutQuorumWait()} + * acquires the monitor. The pre-existing {@code if (completed) return;} guard does not + * cover this because recycling resets {@code completed} to {@code false}; without an + * explicit null check the method NPEs on {@code lh.getLedgerMetadata()} (or similar) + * and, worse, may invoke {@code handleUnrecoverableErrorDuringAdd} on a stale handle. + */ + @Test + public void testTimeoutQuorumWaitIsNoOpWhenAlreadyRecycled() { + PendingAddOp op = PendingAddOp.create( + lh, mockClientContext, lh.getCurrentEnsemble(), + payload, WriteFlag.NONE, + (rc, handle, entryId, qwcLatency, ctx) -> {}, null); + + // Simulate post-recycle state: recyclePendAddOpObject() has already nulled these fields. + op.lh = null; + op.clientCtx = null; + + // Must not throw NPE — and must not invoke the unrecoverable-error handler on the + // (now-stale) ledger handle, which would cause spurious failures on a recycled op. + op.timeoutQuorumWait(); + verify(lh, never()).handleUnrecoverableErrorDuringAdd(anyInt()); + } }
