[FLINK-30476] TrackingFsDataInputStream batch tracking issue

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.15.1, 1.15.2, 1.15.3
• Component/s: Connectors / FileSystem
• Labels: pull-request-available stale-assigned

Language

java

• Assignee: Denis Golovachev
• Reporter: Denis Golovachev
• Created: 21/Dec/22 12:57
• Updated: 14/Aug/23 10:35

Description

org.apache.flink.connector.file.src.impl.StreamFormatAdapter.TrackingFsDataInputStream wraps underlying InputStream to count bytes consumed.
org.apache.flink.connector.file.src.impl.StreamFormatAdapter.Reader relies on this to create batches of data.

            while (stream.hasRemainingInBatch() && (next = reader.read()) != null) {
                result.add(next);
            }

org.apache.flink.connector.file.src.impl.StreamFormatAdapter.TrackingFsDataInputStream#read(byte[], int, int) contains a bug that can lead to arbitrary size batches due to counter (remainingInBatch) underflow.

        public int read(byte[] b, int off, int len) throws IOException {
            remainingInBatch -= len;
            return stream.read(b, off, len);
        }

Every time we perform a stream.read() it may return less than len according to the javadoc.

Params:
b – the buffer into which the data is read. off – the start offset in array b at which the data is written. len – the maximum number of bytes to read.
Returns:
the total number of bytes read into the buffer, or -1 if there is no more data because the end of the stream has been reached.

But current implementation accounts only bytes that were requested ({}{{len{}}}).

E.g. S3 Hadoop FS can return less than len as a result of stream.read(b, off, len). This is expected and readers are aware of this
org.apache.parquet.io.DelegatingSeekableInputStream#readFully(java.io.InputStream, byte[], int, int)

As a result reading parquet file may result in underflow TrackingFsDataInputStream#read(byte[], int, int) because parquet reader tries to read the whole Row Group (large) and may execute read() multiple times. Underflow leads to unlimited batch size that may lead to OOM.

Issue Links

• links to: GitHub Pull Request #21544