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) <noreply@anthropic.com>

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;
-    volatile ClientContext clientCtx;
+    ClientContext clientCtx;
     boolean isRecoveryAdd = false;
     volatile long requestTimeNanos;
     long qwcLatency; // Quorum Write Completion Latency after response from quorum bookies.
@@ -154,21 +154,28 @@ 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.
+        // 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) >= clientCtx.getConf().addEntryQuorumTimeoutNanos) {
+        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

@@ -24,7 +24,10 @@
 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;
@@ -165,4 +168,33 @@ public void testMaybeTimeoutReturnsTrueWhenQuorumTimeoutExpired() {
 
         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());
+    }
 }