Issue 4503: Added check into BufferedChannel's read to avoid endless loop

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java

@@ -243,6 +243,11 @@ public long forceWrite(boolean forceMetadata) throws IOException {
 
     @Override
     public synchronized int read(ByteBuf dest, long pos, int length) throws IOException {
+        if (dest.writableBytes() < length) {
+            throw new IllegalArgumentException("dest buffer remaining capacity is not enough"
+                    + "(must be at least as \"length\"=" + length + ")");
+        }
+
         long prevPos = pos;
         while (length > 0) {
             // check if it is in the write buffer
@@ -295,4 +300,4 @@ public synchronized int getNumOfBytesInWriteBuffer() {
     long getUnpersistedBytes() {
         return unpersistedBytes.get();
     }
-}
+}

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java

@@ -21,10 +21,13 @@
 
 package org.apache.bookkeeper.bookie;
 
+import static org.junit.Assert.assertThrows;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import java.io.File;
+import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.nio.channels.FileChannel;
 import java.util.Random;
@@ -126,6 +129,41 @@ public void testBufferedChannel(int byteBufLength, int numOfWrites, int unpersis
         fileChannel.close();
     }
 
+    @Test
+    public void testBufferedChannelReadWhenDestBufSizeExceedsReadLength() throws IOException {
+        doTestBufferedChannelReadThrowing(100, 60);
+    }
+
+    @Test
+    public void testBufferedChannelReadWhenDestBufSizeDoesNotExceedReadLength() throws IOException {
+        doTestBufferedChannelReadThrowing(100, 110);
+    }
+
+    private void doTestBufferedChannelReadThrowing(int destBufSize, int readLength) throws IOException {
+        File newLogFile = File.createTempFile("test", "log");
+        newLogFile.deleteOnExit();
+
+        try (RandomAccessFile raf = new RandomAccessFile(newLogFile, "rw")) {
+            FileChannel fileChannel = raf.getChannel();
+
+            try (BufferedChannel bufferedChannel = new BufferedChannel(
+                UnpooledByteBufAllocator.DEFAULT, fileChannel,
+                INTERNAL_BUFFER_WRITE_CAPACITY, INTERNAL_BUFFER_READ_CAPACITY, 0)) {
+
+                bufferedChannel.write(generateEntry(500));
+
+                ByteBuf destBuf = UnpooledByteBufAllocator.DEFAULT.buffer(destBufSize);
+
+                if (destBufSize < readLength) {
+                    assertThrows(IllegalArgumentException.class,
+                        () -> bufferedChannel.read(destBuf, 0, readLength));
+                } else {
+                    bufferedChannel.read(destBuf, 0, readLength);
+                }
+            }
+        }
+    }
+
     private static ByteBuf generateEntry(int length) {
         byte[] data = new byte[length];
         ByteBuf bb = Unpooled.buffer(length);