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 6b82200b3fa..7ab8da1d49f 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 @@ private void sendWriteRequest(List ensemble, int bookieIndex) { } 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 @@ boolean maybeTimeout() { } 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 5fb318c51fe..968bbaab565 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 @@ 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 void testExecuteAfterCancelled() { assertNull(op.lh); } + /** + * Verify that {@link PendingAddOp#maybeTimeout()} returns {@code false} without throwing + * {@link NullPointerException} when {@code clientCtx} is {@code null}. + * + *

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}). + * + *

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()); + } }