batchReadEntries runs indefinitely with batchRead disabled

[mazzucchi-andrea]

BUG REPORT

In batchReadEntries using Wire Protocol V2 with batch read enabled, if the maxCount parameter is set to 0, the method returns all entries that can fit within the maxSize limit. However, without batch read enabled, maxSize is not considered, and the method should, theoretically, return all entries on the ledger. Nevertheless, the test results in a Timeout Failure.

To Reproduce

I wrote a simple test based on the existing tests in BookKeeperTest.java

package org.apache.bookkeeper.client;

import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;

import java.util.Enumeration;

import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

public class BatchReadEntriesMaxCountTest extends BookKeeperClusterTestCase {
    private final BookKeeper.DigestType digestType;

    public BatchReadEntriesMaxCountTest() {
        super(3);
        this.digestType = BookKeeper.DigestType.CRC32;
    }

    @Test
    @Timeout(120)
    public void testBatchReadMaxCountV3Protocol() throws Exception {
        ClientConfiguration conf = new ClientConfiguration();
        conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
        conf.setBatchReadEnabled(false);
        int numEntries = 100;
        byte[] data = "foobar".getBytes();
        try (BookKeeper bkc = new BookKeeper(conf)) {
            long ledgerId;
            try (LedgerHandle lh = bkc.createLedger(2, 2, digestType, "testPasswd".getBytes())) {
                ledgerId = lh.getId();
                for (int i = 0; i < numEntries; i++) {
                    assertTrue(lh.addEntry(data) >= 0);
                }
            }
            try (LedgerHandle lh = bkc.openLedger(ledgerId, digestType, "testPasswd".getBytes())) {
                assertEquals(numEntries - 1, lh.readLastConfirmed());
                int entries = 0;
                for (Enumeration<LedgerEntry> readEntries = lh.batchReadEntries(0, numEntries, 5 * 1024 * 1024);
                     readEntries.hasMoreElements();) {
                    LedgerEntry entry = readEntries.nextElement();
                    assertArrayEquals(data, entry.getEntry());
                    entries++;
                }
                assertEquals(numEntries, entries);

                //The maxCount is 0, the result is not limited by maxSize if batchRead is disabled.
                entries = 0;
                for (Enumeration<LedgerEntry> readEntries = lh.batchReadEntries(0, 0, 5 * 1024 * 1024);
                     readEntries.hasMoreElements();) {
                    LedgerEntry entry = readEntries.nextElement();
                    assertArrayEquals(data, entry.getEntry());
                    entries++;
                }
                assertEquals(numEntries, entries);

                //The maxCount is -1, should throw an BKIncorrectParameterException.
                assertThrows(BKException.BKIncorrectParameterException.class,
                        () -> lh.batchReadEntries(0, -1, 5 * 1024 * 1024));

            }
        }
    }
}

Expected behavior

With batchRead disabled:

• if maxCount is 0, batchReadEntries should return all entries on the Ledger without being limited by maxSize.
• if maxCount is -1, batchReadEntries should throw BKIncorrectParameterException.

[mazzucchi-andrea]

@gulyx