BufferedChannel's read(ByteBuf, long, int) gets stuck in loop

[StefanoBelli]

BUG REPORT

Describe the bug

While testing BufferedChannel's read instance method (with buf capacity=100B, wrote data through the buffered channel=300B), it happens that when the read's formal parameter "length" is greater than 256, read gets stuck in a loop for some reason.

So if i want to read from pos=43, length=256 then test succeeds, but for pos=43, length=257 we have an endless loop.

Exchanging parameters actual values (from pos=43, length=257 to pos=256, length=43, considering that pos is an index) results in a good test outcome.

To Reproduce

Run this test (emplace under bookkeeper-server/src/test/java/...): parameterized JUnit runner allows to run same test with different params: the publicly-available static method "params()" (annotated with "Parameters") in BufferedChannelTest provides those params - in our case those params are pos and length being passed to the read() function of BufferedChannel (see test method). Commented out params are the one that fails, above each one there are the same exchanged values (which works for some reason)

package org.apache.bookkeeper.bookie;

import static org.junit.Assert.assertEquals;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Random;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

@RunWith(Parameterized.class)
public class BufferedChannelTest {

    private static final String WRAPPED_FILE_PATH = System.getProperty("user.dir") + "/bktest.txt";
    private static final int BUFFERED_CHANNEL_CAPACITY = 100;
    private static final int RANDOM_DATA_LENGTH = 300;

    private final int readPos;
    private final int readLen;

    @Parameterized.Parameters
    public static Iterable<Object[]> params() {
        return Arrays.asList(
                new Object[][] {
                        { 10, 20 },
                        { 30, 40 },
                        { 70, 230 },
                        { 50, 200 },
                        { 43, 256 },

                        { 256, 43 },
                        //{ 43, 257 },

                        { 289, 10 },
                        //{ 9, 290 },

                        { 299, 0 },
                        //{ 0, 300 },

                        { 259, 40 },
                        //{ 40, 260 },

                        { 259, 30 },
                        //{ 30, 260 },
                }
        );
    }

    public BufferedChannelTest(int readPos, int readLen) {
        this.readPos = readPos;
        this.readLen = readLen;
    }

    @Test
    public void test() throws IOException {
        ByteBuf destBuf = allocBuf();
        String randomData = generateString(RANDOM_DATA_LENGTH);

        try(RandomAccessFile raf = new RandomAccessFile(WRAPPED_FILE_PATH, "rw")) {
            try (BufferedChannel bufChannel = new BufferedChannel(
                    ByteBufAllocator.DEFAULT,
                    raf.getChannel(),
                    BUFFERED_CHANNEL_CAPACITY)) {
                bufChannel.write(allocBuf(randomData));
                bufChannel.read(destBuf, readPos, readLen);
            }
        }

        assertEquals(
                randomData.substring(readPos, readPos + readLen),
                bufContentToString(readLen, destBuf));
    }

    protected static ByteBuf allocBuf(String data) {
        byte[] b = data.getBytes(StandardCharsets.UTF_8);
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
        buf.writeBytes(b);

        return buf;
    }

    protected static ByteBuf allocBuf() {
        return ByteBufAllocator.DEFAULT.buffer();
    }

    protected static String bufContentToString(int size, ByteBuf buf) {
        byte[] b = new byte[size];
        buf.getBytes(0, b);

        return new String(b);
    }

    protected static String concatArray(String[] strs) {
        StringBuilder builder = new StringBuilder();

        for(final String str : strs) {
            builder.append(str);
        }

        return builder.toString();
    }

    protected static String generateString(int len) {
        StringBuilder s = new StringBuilder();

        for(int i = 1; i <= len; ++i) {
            s.append(randomChar());
        }

        return s.toString();
    }

    protected static char randomChar() {
        final Random random = new Random();
        final String alphabet = "qwertyuiopasdfghjklzxcvbnm1234567890";

        return alphabet.charAt(random.nextInt(alphabet.length()));
    }
}

Expected behavior

Test assertion should pass anyway, with any combination of pos, length