Fill the lookahead buffer on short reads from a chunked source

src/main/java/org/apache/commons/csv/ExtendedBufferedReader.java

@@ -225,6 +225,37 @@ public void mark(final int readAheadLimit) throws IOException {
         super.mark(readAheadLimit);
     }
 
+    /**
+     * Fills {@code array} with the characters that follow the current position without consuming them.
+     * <p>
+     * Overridden because the inherited implementation stops at the first short read, which leaves the tail of the array holding stale content when the source
+     * delivers data in chunks. Callers compare the whole array against a multi-character delimiter, so a partial fill makes them miss a delimiter that is
+     * really there.
+     * </p>
+     *
+     * @param array the buffer to fill.
+     * @return the number of characters peeked, or {@link IOUtils#EOF} at the end of the stream.
+     * @throws IOException If an I/O error occurs.
+     */
+    @Override
+    public int peek(final char[] array) throws IOException {
+        final int length = array.length;
+        if (length == 0) {
+            return 0;
+        }
+        super.mark(length);
+        int len = 0;
+        while (len < length) {
+            final int more = super.read(array, len, length - len);
+            if (more == EOF) {
+                break;
+            }
+            len += more;
+        }
+        super.reset();
+        return len == 0 ? EOF : len;
+    }
+
     @Override
     public int read() throws IOException {
         final int current = super.read();
@@ -244,7 +275,16 @@ public int read(final char[] buf, final int offset, final int length) throws IOE
         if (length == 0) {
             return 0;
         }
-        final int len = super.read(buf, offset, length);
+        int len = super.read(buf, offset, length);
+        // The underlying buffered reader stops early once the source reports it is not ready, so a stream that delivers data in chunks (a socket or a pipe)
+        // yields a short read. Callers match multi-character sequences against this buffer, so keep reading until it is full or the source is exhausted.
+        while (len > 0 && len < length) {
+            final int more = super.read(buf, offset + len, length - len);
+            if (more == EOF) {
+                break;
+            }
+            len += more;
+        }
         if (encoder != null && len > 0) {
             this.bytesRead += getEncodedCharLength(buf, offset, len);
         }

src/test/java/org/apache/commons/csv/CSVParserTest.java

@@ -1709,6 +1709,47 @@ void testParseWithDelimiterStringWithEscape() throws IOException {
         }
     }
 
+    @Test
+    void testParseWithDelimiterStringFromChunkedReader() throws IOException {
+        // A reader that hands out one character at a time and never reports itself ready, like a socket or pipe.
+        final Reader chunked = new Reader() {
+
+            private final String content = "a[|]b\r\nc![!|!]d[|]e";
+            private int index;
+
+            @Override
+            public void close() {
+                // nothing to close
+            }
+
+            @Override
+            public boolean ready() {
+                return false;
+            }
+
+            @Override
+            public int read(final char[] buf, final int offset, final int length) {
+                if (index >= content.length()) {
+                    return -1;
+                }
+                if (length <= 0) {
+                    return 0;
+                }
+                buf[offset] = content.charAt(index++);
+                return 1;
+            }
+        };
+        final CSVFormat csvFormat = CSVFormat.DEFAULT.builder().setDelimiter("[|]").setEscape('!').get();
+        try (CSVParser csvParser = csvFormat.parse(chunked)) {
+            CSVRecord csvRecord = csvParser.nextRecord();
+            assertEquals("a", csvRecord.get(0));
+            assertEquals("b", csvRecord.get(1));
+            csvRecord = csvParser.nextRecord();
+            assertEquals("c[|]d", csvRecord.get(0));
+            assertEquals("e", csvRecord.get(1));
+        }
+    }
+
     @Test
     void testParseWithDelimiterStringWithQuote() throws IOException {
         final String source = "'a[|]b[|]c'[|]xyz\r\nabc[abc][|]xyz";

src/test/java/org/apache/commons/csv/ExtendedBufferedReaderTest.java

@@ -240,4 +240,72 @@ void testReadLookahead2() throws Exception {
             assertEquals('d', br.getLastChar());
         }
     }
+
+    @Test
+    void testReadAndPeekArrayFromChunkedReader() throws Exception {
+        try (ExtendedBufferedReader br = new ExtendedBufferedReader(new ChunkedReader("abcdef"))) {
+            final char[] peeked = new char[3];
+            assertEquals(3, br.peek(peeked));
+            assertArrayEquals(new char[] { 'a', 'b', 'c' }, peeked);
+            final char[] read = new char[3];
+            assertEquals(3, br.read(read, 0, 3));
+            assertArrayEquals(new char[] { 'a', 'b', 'c' }, read);
+        }
+    }
+
+    @Test
+    void testReadArrayPastEndOfChunkedReader() throws Exception {
+        try (ExtendedBufferedReader br = new ExtendedBufferedReader(new ChunkedReader("ab"))) {
+            final char[] read = new char[4];
+            assertEquals(2, br.read(read, 0, 4));
+            assertEquals('a', read[0]);
+            assertEquals('b', read[1]);
+            assertEquals(EOF, br.read(read, 0, 4));
+        }
+    }
+
+    @Test
+    void testPeekArrayPastEndOfChunkedReader() throws Exception {
+        try (ExtendedBufferedReader br = new ExtendedBufferedReader(new ChunkedReader("ab"))) {
+            final char[] peeked = new char[4];
+            assertEquals(2, br.peek(peeked));
+            assertEquals('a', peeked[0]);
+            assertEquals('b', peeked[1]);
+        }
+    }
+
+    /**
+     * A reader that returns one character per call and never reports itself ready, like a socket or pipe that delivers data in chunks.
+     */
+    private static final class ChunkedReader extends java.io.Reader {
+
+        private final String content;
+        private int index;
+
+        ChunkedReader(final String content) {
+            this.content = content;
+        }
+
+        @Override
+        public void close() {
+            // nothing to close
+        }
+
+        @Override
+        public boolean ready() {
+            return false;
+        }
+
+        @Override
+        public int read(final char[] buf, final int offset, final int length) {
+            if (index >= content.length()) {
+                return EOF;
+            }
+            if (length <= 0) {
+                return 0;
+            }
+            buf[offset] = content.charAt(index++);
+            return 1;
+        }
+    }
 }