From caf0a6c887d89214638ebc509cc831aa333e16b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Mon, 24 Aug 2026 13:25:05 +0200 Subject: [PATCH] AVRO-4048: [java] Handle InputStream.skip() robustly in binary decoders The binary decoders mishandled InputStream.skip() returning 0. Per its contract, skip() may return 0 without being at end of stream, and negative returns are not specified. The previous code treated two consecutive 0 returns (and any negative return) as EOF, which could raise a spurious EOFException on streams that legitimately return 0 from skip(). It also carried a dead "negative return" branch. Additionally, BinaryDecoder.InputStreamByteSource.trySkipBytes requested the original full length on every iteration (in.skip(length)) instead of the remaining amount (in.skip(leftToSkip)). On a partial skip this re-requested too much and could advance the stream past the intended position (over-skip / stream corruption). Replace the fragile heuristics with a single-byte read() probe: when skip() returns a non-positive value, read one byte to distinguish a genuine EOF (read() == -1) from a transient inability to skip, then continue. Fix trySkipBytes to skip only the remaining count. The same read()-probe logic is applied to DirectBinaryDecoder.doSkipBytes, which previously treated skip() == 0 as immediate EOF. Adds TestBinaryDecoderSkip covering: no spurious EOF on a zero-skip stream (buffered and direct), genuine EOF still detected when skipping past the end, and no over-skip on partial-skip streams. --- .../org/apache/avro/io/BinaryDecoder.java | 48 ++---- .../apache/avro/io/DirectBinaryDecoder.java | 11 +- .../apache/avro/io/TestBinaryDecoderSkip.java | 150 ++++++++++++++++++ 3 files changed, 175 insertions(+), 34 deletions(-) create mode 100644 lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoderSkip.java diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java index 3974eb5c621..fad7d09db1e 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java @@ -870,28 +870,22 @@ private InputStreamByteSource(InputStream in) { @Override protected void skipSourceBytes(long length) throws IOException { - boolean readZero = false; while (length > 0) { long n = in.skip(length); if (n > 0) { length -= n; continue; } - // The inputStream contract is evil. - // zero "might" mean EOF. So check for 2 in a row, we will - // infinite loop waiting for -1 with some classes others - // spuriously will return 0 on occasion without EOF - if (n == 0) { - if (readZero) { - isEof = true; - throw new EOFException(); - } - readZero = true; - continue; + // InputStream.skip() may return 0 without being at end of stream (its + // contract explicitly permits it), and is not specified to return a + // negative value. Rather than looping forever or treating a spurious 0 + // as EOF, probe with a single read() to tell a real EOF apart from a + // transient inability to skip. + if (in.read() < 0) { + isEof = true; + throw new EOFException(); } - // read negative - isEof = true; - throw new EOFException(); + length--; } } @@ -899,29 +893,19 @@ protected void skipSourceBytes(long length) throws IOException { protected long trySkipBytes(long length) throws IOException { long leftToSkip = length; try { - boolean readZero = false; while (leftToSkip > 0) { - long n = in.skip(length); + long n = in.skip(leftToSkip); if (n > 0) { leftToSkip -= n; continue; } - // The inputStream contract is evil. - // zero "might" mean EOF. So check for 2 in a row, we will - // infinite loop waiting for -1 with some classes others - // spuriously will return 0 on occasion without EOF - if (n == 0) { - if (readZero) { - isEof = true; - break; - } - readZero = true; - continue; + // See skipSourceBytes: distinguish a real EOF from a transient skip() + // returning 0 by probing with a single read(). + if (in.read() < 0) { + isEof = true; + break; } - // read negative - isEof = true; - break; - + leftToSkip--; } } catch (EOFException eof) { isEof = true; diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/DirectBinaryDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/DirectBinaryDecoder.java index ac251550da2..c2c52668bec 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/DirectBinaryDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/DirectBinaryDecoder.java @@ -163,10 +163,17 @@ public ByteBuffer readBytes(ByteBuffer old) throws IOException { protected void doSkipBytes(long length) throws IOException { while (length > 0) { long n = in.skip(length); - if (n <= 0) { + if (n > 0) { + length -= n; + continue; + } + // InputStream.skip() may return 0 without being at end of stream, so probe + // with a single read() to distinguish a genuine EOF from a transient 0 + // instead of failing spuriously. + if (in.read() < 0) { throw new EOFException(); } - length -= n; + length--; } } diff --git a/lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoderSkip.java b/lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoderSkip.java new file mode 100644 index 00000000000..d5fa0c79e19 --- /dev/null +++ b/lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoderSkip.java @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.avro.io; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.ByteArrayInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; + +import org.junit.jupiter.api.Test; + +/** + * Regression tests for AVRO-4253's sibling AVRO-4048: robust handling of + * {@link InputStream#skip(long)} in the binary decoders. + *

+ * {@code skip()} is allowed to return {@code 0} without being at end of stream, + * and negative returns are not part of its contract. The decoders must not treat + * a transient {@code 0} as EOF, and must never skip more bytes than requested. + */ +public class TestBinaryDecoderSkip { + + /** An InputStream whose skip() always returns 0, forcing the read() fallback. */ + private static final class ZeroSkipInputStream extends InputStream { + private final ByteArrayInputStream delegate; + + ZeroSkipInputStream(byte[] data) { + this.delegate = new ByteArrayInputStream(data); + } + + @Override + public int read() throws IOException { + return delegate.read(); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + return delegate.read(b, off, len); + } + + @Override + public long skip(long n) { + // Simulate a stream that never manages to skip, even when not at EOF. + return 0; + } + } + + /** An InputStream whose skip() advances by at most {@code chunk} bytes per call. */ + private static final class PartialSkipInputStream extends InputStream { + private final ByteArrayInputStream delegate; + private final long chunk; + + PartialSkipInputStream(byte[] data, long chunk) { + this.delegate = new ByteArrayInputStream(data); + this.chunk = chunk; + } + + @Override + public int read() throws IOException { + return delegate.read(); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + return delegate.read(b, off, len); + } + + @Override + public long skip(long n) { + return delegate.skip(Math.min(n, chunk)); + } + } + + private static byte[] ramp(int len) { + byte[] data = new byte[len]; + for (int i = 0; i < len; i++) { + data[i] = (byte) i; + } + return data; + } + + /** + * A skip() that returns 0 without EOF must not cause a spurious EOFException, + * and the decoder must be positioned exactly after the skipped bytes. + */ + @Test + public void bufferedSkipFixedWithZeroSkipStream() throws IOException { + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(new ZeroSkipInputStream(ramp(10)), null); + decoder.skipFixed(4); + byte[] rest = new byte[6]; + decoder.readFixed(rest, 0, 6); + assertArrayEquals(new byte[] { 4, 5, 6, 7, 8, 9 }, rest); + } + + @Test + public void directSkipFixedWithZeroSkipStream() throws IOException { + BinaryDecoder decoder = DecoderFactory.get().directBinaryDecoder(new ZeroSkipInputStream(ramp(10)), null); + decoder.skipFixed(4); + byte[] rest = new byte[6]; + decoder.readFixed(rest, 0, 6); + assertArrayEquals(new byte[] { 4, 5, 6, 7, 8, 9 }, rest); + } + + /** Skipping past the real end of stream must still raise EOFException. */ + @Test + public void bufferedSkipFixedPastEndThrows() { + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(new ZeroSkipInputStream(ramp(3)), null); + assertThrows(EOFException.class, () -> decoder.skipFixed(10)); + } + + @Test + public void directSkipFixedPastEndThrows() { + BinaryDecoder decoder = DecoderFactory.get().directBinaryDecoder(new ZeroSkipInputStream(ramp(3)), null); + assertThrows(EOFException.class, () -> decoder.skipFixed(10)); + } + + /** + * When skip() returns fewer bytes than requested, the decoder must request + * only the remaining count on subsequent calls; otherwise it over-skips and + * corrupts the stream position (AVRO-4048). + */ + @Test + public void inputStreamSkipDoesNotOverSkip() throws IOException { + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(new PartialSkipInputStream(ramp(10), 3), null); + InputStream is = decoder.inputStream(); + + long skipped = is.skip(5); + assertEquals(5, skipped, "must skip exactly the requested number of bytes"); + // The next byte must be data[5]; an over-skip would surface data[6]. + assertEquals(5, is.read(), "stream must be positioned exactly after the skipped bytes"); + } +}