Revert "Fix read thread blocking in sendResponseAndWait causing READ_ENTRY_REQUEST p99 latency spike (#4730)" (#4830)

This reverts commit 8664dd9.

Author: hangc0276

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java

@@ -1078,16 +1078,10 @@ public ServerConfiguration setMaxAddsInProgressLimit(int value) {
     /**
      * Get max number of reads in progress. 0 == unlimited.
      *
-     * <p>This limit bounds the memory used by read responses that have been read from storage
-     * but not yet flushed to the network. Since read response writes are non-blocking,
-     * without this limit a slow consumer could cause unbounded memory growth.
-     * The default value of 10000 provides a reasonable balance between throughput and memory usage.
-     * Tune based on your average entry size: memoryBudget / avgEntrySize.
-     *
      * @return Max number of reads in progress.
      */
     public int getMaxReadsInProgressLimit() {
-        return this.getInt(MAX_READS_IN_PROGRESS_LIMIT, 10000);
+        return this.getInt(MAX_READS_IN_PROGRESS_LIMIT, 0);
     }
 
     /**

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PacketProcessorBase.java

@@ -19,8 +19,8 @@
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelPromise;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import lombok.CustomLog;
 import org.apache.bookkeeper.common.util.MathUtils;
@@ -73,12 +73,10 @@ protected void sendWriteReqResponse(int rc, Object response, OpStatsLogger stats
     protected void sendReadReqResponse(int rc, Object response, OpStatsLogger statsLogger, boolean throttle) {
         if (throttle) {
             sendResponseAndWait(rc, response, statsLogger);
-            // onReadRequestFinish is called asynchronously in the ChannelFutureListener
-            // inside sendResponseAndWait to maintain throttling without blocking the thread.
         } else {
             sendResponse(rc, response, statsLogger);
-            requestProcessor.onReadRequestFinish();
         }
+        requestProcessor.onReadRequestFinish();
     }
 
     protected void sendResponse(int rc, Object response, OpStatsLogger statsLogger) {
@@ -146,42 +144,27 @@ protected void sendResponse(int rc, Object response, OpStatsLogger statsLogger)
     }
 
     /**
-     * Write on the channel and notify completion via a listener.
+     * Write on the channel and wait until the write is completed.
      *
-     * <p>This provides auto-throttling by holding the read semaphore until the write completes,
-     * without blocking the read thread pool thread. The read thread is freed immediately to
-     * process other requests, while the semaphore prevents unbounded read concurrency.
+     * <p>That will make the thread to get blocked until we're able to
+     * write everything on the TCP stack, providing auto-throttling
+     * and avoiding using too much memory when handling read-requests.
      */
     protected void sendResponseAndWait(int rc, Object response, OpStatsLogger statsLogger) {
-        // Capture fields before the processor may be recycled after this method returns.
-        final long capturedEnqueueNanos = this.enqueueNanos;
-        final BookieRequestProcessor processor = this.requestProcessor;
         try {
             Channel channel = requestHandler.ctx().channel();
             ChannelFuture future = channel.writeAndFlush(response);
-            future.addListener((ChannelFutureListener) f -> {
-                if (!f.isSuccess()) {
-                    log.debug().exception(f.cause()).log("Netty channel write exception");
-                }
-                if (BookieProtocol.EOK == rc) {
-                    statsLogger.registerSuccessfulEvent(
-                            MathUtils.elapsedNanos(capturedEnqueueNanos), TimeUnit.NANOSECONDS);
-                } else {
-                    statsLogger.registerFailedEvent(
-                            MathUtils.elapsedNanos(capturedEnqueueNanos), TimeUnit.NANOSECONDS);
-                }
-                processor.onReadRequestFinish();
-            });
-        } catch (Exception e) {
-            log.debug().exception(e).log("Netty channel write exception");
-            if (BookieProtocol.EOK == rc) {
-                statsLogger.registerSuccessfulEvent(
-                        MathUtils.elapsedNanos(capturedEnqueueNanos), TimeUnit.NANOSECONDS);
-            } else {
-                statsLogger.registerFailedEvent(
-                        MathUtils.elapsedNanos(capturedEnqueueNanos), TimeUnit.NANOSECONDS);
+            if (!channel.eventLoop().inEventLoop()) {
+                future.get();
             }
-            processor.onReadRequestFinish();
+        } catch (ExecutionException | InterruptedException e) {
+            log.debug().exception(e).log("Netty channel write exception");
+            return;
+        }
+        if (BookieProtocol.EOK == rc) {
+            statsLogger.registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
+        } else {
+            statsLogger.registerFailedEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
         }
     }
 

bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/ReadEntryProcessorTest.java

@@ -19,14 +19,12 @@
 package org.apache.bookkeeper.proto;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.RETURNS_SELF;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -41,12 +39,10 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.bookkeeper.bookie.Bookie;
 import org.apache.bookkeeper.bookie.BookieException;
 import org.apache.bookkeeper.common.concurrent.FutureUtils;
-import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.proto.BookieProtocol.ReadRequest;
 import org.apache.bookkeeper.proto.BookieProtocol.Response;
 import org.apache.bookkeeper.stats.NullStatsLogger;
@@ -201,178 +197,4 @@ public void testNonFenceRequest() throws Exception {
         assertEquals(BookieProtocol.READENTRY, response.getOpCode());
         assertEquals(BookieProtocol.EOK, response.getErrorCode());
     }
-
-    /**
-     * Test that when throttleReadResponses=true and the caller is not in the Netty event loop,
-     * the read thread is not blocked by the write. onReadRequestFinish() should only be called
-     * after the write future completes, preserving throttling without blocking the thread.
-     */
-    @Test
-    public void testThrottledReadNonBlockingOnSuccess() throws Exception {
-        // Setup event loop to simulate read worker thread (not event loop thread)
-        EventLoop eventLoop = mock(EventLoop.class);
-        when(eventLoop.inEventLoop()).thenReturn(false);
-        doAnswer(inv -> {
-            ((Runnable) inv.getArgument(0)).run();
-            return null;
-        }).when(eventLoop).execute(any(Runnable.class));
-        when(channel.eventLoop()).thenReturn(eventLoop);
-
-        // Use a controllable promise so we can verify deferred behavior
-        DefaultChannelPromise writeFuture = new DefaultChannelPromise(channel);
-        doAnswer(inv -> writeFuture).when(channel).writeAndFlush(any(Response.class));
-
-        long ledgerId = System.currentTimeMillis();
-        ReadRequest request = ReadRequest.create(
-                BookieProtocol.CURRENT_PROTOCOL_VERSION, ledgerId, 1, (short) 0, new byte[]{});
-        ReadEntryProcessor processor = ReadEntryProcessor.create(
-                request, requestHandler, requestProcessor, null, true /* throttle */);
-
-        // run() should return immediately without blocking on the write
-        processor.run();
-
-        // Write should have been issued
-        verify(channel, times(1)).writeAndFlush(any(Response.class));
-        // But onReadRequestFinish should NOT have been called yet — write not completed
-        verify(requestProcessor, never()).onReadRequestFinish();
-
-        // Complete the write
-        writeFuture.setSuccess();
-
-        // Now onReadRequestFinish should have been called
-        verify(requestProcessor, times(1)).onReadRequestFinish();
-    }
-
-    /**
-     * Test that onReadRequestFinish() is still called even when the write fails,
-     * so the read semaphore is always released.
-     */
-    @Test
-    public void testThrottledReadNonBlockingOnWriteFailure() throws Exception {
-        EventLoop eventLoop = mock(EventLoop.class);
-        when(eventLoop.inEventLoop()).thenReturn(false);
-        doAnswer(inv -> {
-            ((Runnable) inv.getArgument(0)).run();
-            return null;
-        }).when(eventLoop).execute(any(Runnable.class));
-        when(channel.eventLoop()).thenReturn(eventLoop);
-
-        DefaultChannelPromise writeFuture = new DefaultChannelPromise(channel);
-        doAnswer(inv -> writeFuture).when(channel).writeAndFlush(any(Response.class));
-
-        long ledgerId = System.currentTimeMillis();
-        ReadRequest request = ReadRequest.create(
-                BookieProtocol.CURRENT_PROTOCOL_VERSION, ledgerId, 1, (short) 0, new byte[]{});
-        ReadEntryProcessor processor = ReadEntryProcessor.create(
-                request, requestHandler, requestProcessor, null, true /* throttle */);
-
-        processor.run();
-
-        verify(channel, times(1)).writeAndFlush(any(Response.class));
-        verify(requestProcessor, never()).onReadRequestFinish();
-
-        // Fail the write
-        writeFuture.setFailure(new IOException("channel write failed"));
-
-        // onReadRequestFinish must still be called to release the read semaphore
-        verify(requestProcessor, times(1)).onReadRequestFinish();
-    }
-
-    /**
-     * Test that when throttleReadResponses=false, onReadRequestFinish() is called
-     * synchronously before run() returns.
-     */
-    @Test
-    public void testNonThrottledReadCallsOnFinishSynchronously() throws Exception {
-        // sendResponse (non-throttle path) uses channel.isActive() and two-arg writeAndFlush
-        when(channel.isActive()).thenReturn(true);
-        when(channel.writeAndFlush(any(), any(ChannelPromise.class))).thenReturn(mock(ChannelPromise.class));
-
-        long ledgerId = System.currentTimeMillis();
-        ReadRequest request = ReadRequest.create(
-                BookieProtocol.CURRENT_PROTOCOL_VERSION, ledgerId, 1, (short) 0, new byte[]{});
-        ReadEntryProcessor processor = ReadEntryProcessor.create(
-                request, requestHandler, requestProcessor, null, false /* no throttle */);
-
-        processor.run();
-
-        verify(channel, times(1)).writeAndFlush(any(), any(ChannelPromise.class));
-        // onReadRequestFinish should have been called synchronously
-        verify(requestProcessor, times(1)).onReadRequestFinish();
-    }
-
-    /**
-     * Verify that maxReadsInProgressLimit defaults to 10000 (enabled),
-     * ensuring non-blocking read response writes are bounded by default.
-     */
-    @Test
-    public void testDefaultMaxReadsInProgressLimitIsEnabled() {
-        ServerConfiguration conf = new ServerConfiguration();
-        assertEquals("maxReadsInProgressLimit should default to 10000",
-                10000, conf.getMaxReadsInProgressLimit());
-    }
-
-    /**
-     * Test that the read semaphore is held from request creation until the write future completes,
-     * not released when the read thread returns. This ensures that maxReadsInProgressLimit correctly
-     * bounds the number of read responses buffered in memory, even though the read thread is
-     * non-blocking.
-     */
-    @Test
-    public void testThrottledReadHoldsSemaphoreUntilWriteCompletes() throws Exception {
-        // Simulate maxReadsInProgressLimit=1 with a real semaphore
-        Semaphore readsSemaphore = new Semaphore(1);
-
-        doAnswer(inv -> {
-            readsSemaphore.acquireUninterruptibly();
-            return null;
-        }).when(requestProcessor).onReadRequestStart(any(Channel.class));
-        doAnswer(inv -> {
-            readsSemaphore.release();
-            return null;
-        }).when(requestProcessor).onReadRequestFinish();
-
-        // Setup non-event-loop thread
-        EventLoop eventLoop = mock(EventLoop.class);
-        when(eventLoop.inEventLoop()).thenReturn(false);
-        doAnswer(inv -> {
-            ((Runnable) inv.getArgument(0)).run();
-            return null;
-        }).when(eventLoop).execute(any(Runnable.class));
-        when(channel.eventLoop()).thenReturn(eventLoop);
-
-        // Controllable write future
-        DefaultChannelPromise writeFuture = new DefaultChannelPromise(channel);
-        doAnswer(inv -> writeFuture).when(channel).writeAndFlush(any(Response.class));
-
-        long ledgerId = System.currentTimeMillis();
-        ReadRequest request = ReadRequest.create(
-                BookieProtocol.CURRENT_PROTOCOL_VERSION, ledgerId, 1, (short) 0, new byte[]{});
-
-        // create() calls onReadRequestStart → semaphore acquired
-        ReadEntryProcessor processor = ReadEntryProcessor.create(
-                request, requestHandler, requestProcessor, null, true /* throttle */);
-
-        // Semaphore should be acquired (1 permit used)
-        assertEquals("semaphore should have 0 permits after read started",
-                0, readsSemaphore.availablePermits());
-
-        // Run the processor — thread returns immediately (non-blocking)
-        processor.run();
-
-        // Semaphore should STILL be held (write not completed)
-        assertEquals("semaphore should still have 0 permits while write is in progress",
-                0, readsSemaphore.availablePermits());
-
-        // A second read would be unable to acquire the semaphore
-        assertFalse("second read should not be able to acquire semaphore",
-                readsSemaphore.tryAcquire());
-
-        // Complete the write
-        writeFuture.setSuccess();
-
-        // Now semaphore should be released — a new read can proceed
-        assertEquals("semaphore should have 1 permit after write completes",
-                1, readsSemaphore.availablePermits());
-    }
 }