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

[StefanoBelli]

@gulyx

[shoothzj]

I think we can verification and throw an exception while the length of the byte array to be read exceeds the length of the dest Bytebuf, to avoid fall into a dead loop

[StefanoBelli]

A workaround is setting the correct initial capacity for the destination ByteBuf (e.g. in testing class, the method overload allocBuf() should be modified to accept an integer and pass it as initial capacity for ByteBufAllocator.DEFAULT.buffer()).

Having proper checks (even at the beginning of the BufferedChannel.read method) will of course fix this issue definitively, or you may specify this requirement in javadoc (to explicitly set the dest ByteBuf's initial capacity to be greater than or equal to read passed value "length"), or even let BufferedChannel's read to expand the destination ByteBuf object's capacity according to the "length" (e.g. if length > destBufLength then exceed capacity by length - destBufLength) actual value passed to BufferedChannel.read's parameter.

Another valid behaviour would be to partially read until the dest buf is full.

[eolivelli]

I would add the assertion and not change the behavior, as reallocations may lead to unwanted side effects.

This is not a public API and we have control over the usage of the class.

@StefanoBelli would you send a PR to address your problem ?

[StefanoBelli]

Sure, I will within the next few hours

[StefanoBelli]

#4506