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: #4759

Author: eolivelli

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

@@ -66,7 +66,7 @@ class PendingAddOp implements WriteCallback {
     boolean completed = false;
 
     LedgerHandle lh;
-    ClientContext clientCtx;
+    volatile ClientContext clientCtx;
     boolean isRecoveryAdd = false;
     volatile long requestTimeNanos;
     long qwcLatency; // Quorum Write Completion Latency after response from quorum bookies.
@@ -154,6 +154,12 @@ private void sendWriteRequest(List<BookieId> ensemble, int bookieIndex) {
     }
 
     boolean maybeTimeout() {
+        if (clientCtx == null) {
+            // Op has already been recycled: recyclePendAddOpObject() cleared clientCtx while
+            // monitorPendingAddOps() was still holding a reference to it from the iterator.
+            // The add-entry completed before the timeout monitor fired; nothing to time out.
+            return false;
+        }
         if (MathUtils.elapsedNanos(requestTimeNanos) >= clientCtx.getConf().addEntryQuorumTimeoutNanos) {
             timeoutQuorumWait();
             return true;

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java

@@ -20,17 +20,22 @@
 
 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.Mockito.mock;
 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 +92,77 @@ 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());
+    }
 }