Fix NPE in PendingAddOp.maybeTimeout() when clientCtx is null after recycling

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java

@@ -154,15 +154,28 @@ private void sendWriteRequest(List<BookieId> 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;
         }
         return false;
     }
 
     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;
         }
 

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