Core: Fix ByteBufferInputStream.read() to return -1 at EOF (#16167)

* Core: Fix ByteBufferInputStream.read() to return -1 at EOF instead of throwing EOFException (#16127)

* Core: Enhance EOF test coverage for ByteBufferInputStream

Add read(byte[]) assertion to assertAtEOF helper to pin both overloads,
and add testDrainedMultiBufferStream to explicitly exercise the
nextBuffer() -> return -1 code path in MultiBufferInputStream.

Author: sachinnn99

core/src/main/java/org/apache/iceberg/io/MultiBufferInputStream.java

@@ -263,7 +263,7 @@ public int read(byte[] bytes) {
   @Override
   public int read() throws IOException {
     if (current == null) {
-      throw new EOFException();
+      return -1;
     }
 
     while (true) {
@@ -272,7 +272,7 @@ public int read() throws IOException {
         return current.get() & 0xFF; // as unsigned
       } else if (!nextBuffer()) {
         // there are no more buffers
-        throw new EOFException();
+        return -1;
       }
     }
   }

core/src/main/java/org/apache/iceberg/io/SingleBufferInputStream.java

@@ -58,7 +58,7 @@ public long getPos() {
   @Override
   public int read() throws IOException {
     if (!buffer.hasRemaining()) {
-      throw new EOFException();
+      return -1;
     }
     return buffer.get() & 0xFF; // as unsigned
   }

core/src/test/java/org/apache/iceberg/io/TestByteBufferInputStreams.java

@@ -35,6 +35,15 @@ public abstract class TestByteBufferInputStreams {
 
   protected abstract void checkOriginalData();
 
+  private static void assertAtEOF(ByteBufferInputStream stream) throws IOException {
+    long pos = stream.getPos();
+    assertThat(stream.read()).as("read() at EOF").isEqualTo(-1);
+    assertThat(stream.read()).as("read() should keep returning -1 at EOF").isEqualTo(-1);
+    assertThat(stream.read(new byte[1])).as("read(byte[]) at EOF").isEqualTo(-1);
+    assertThat(stream.getPos()).as("Position should not advance past EOF").isEqualTo(pos);
+    assertThat(stream.available()).as("available() should be 0 at EOF").isEqualTo(0);
+  }
+
   @Test
   public void testRead0() throws Exception {
     byte[] bytes = new byte[0];
@@ -67,7 +76,7 @@ public void testReadAll() throws Exception {
 
     assertThat(stream.read(bytes)).as("Should return -1 at end of stream").isEqualTo(-1);
 
-    assertThat(stream.available()).as("Should have no more remaining content").isEqualTo(0);
+    assertAtEOF(stream);
 
     checkOriginalData();
   }
@@ -102,7 +111,7 @@ public void testSmallReads() throws Exception {
 
       assertThat(stream.read(bytes)).as("Should return -1 at end of stream").isEqualTo(-1);
 
-      assertThat(stream.available()).as("Should have no more remaining content").isEqualTo(0);
+      assertAtEOF(stream);
     }
 
     checkOriginalData();
@@ -142,7 +151,7 @@ public void testPartialBufferReads() throws Exception {
 
       assertThat(stream.read(bytes)).as("Should return -1 at end of stream").isEqualTo(-1);
 
-      assertThat(stream.available()).as("Should have no more remaining content").isEqualTo(0);
+      assertAtEOF(stream);
     }
 
     checkOriginalData();
@@ -158,7 +167,7 @@ public void testReadByte() throws Exception {
       assertThat(stream.read()).isEqualTo(i);
     }
 
-    assertThatThrownBy(stream::read).isInstanceOf(EOFException.class).hasMessage(null);
+    assertAtEOF(stream);
 
     checkOriginalData();
   }
@@ -532,4 +541,21 @@ public void testMarkDoubleReset() throws Exception {
         .isInstanceOf(IOException.class)
         .hasMessageStartingWith("No mark defined");
   }
+
+  @Test
+  public void testEmptyStream() throws Exception {
+    assertAtEOF(ByteBufferInputStream.wrap(ByteBuffer.allocate(0)));
+    assertAtEOF(ByteBufferInputStream.wrap(ByteBuffer.allocate(0), ByteBuffer.allocate(0)));
+    assertAtEOF(ByteBufferInputStream.wrap(Collections.emptyList()));
+  }
+
+  @Test
+  public void testDrainedMultiBufferStream() throws Exception {
+    ByteBufferInputStream stream =
+        ByteBufferInputStream.wrap(
+            ByteBuffer.wrap(new byte[] {1, 2, 3}), ByteBuffer.wrap(new byte[] {4, 5}));
+    byte[] buf = new byte[5];
+    assertThat(stream.read(buf)).as("Should read all bytes").isEqualTo(5);
+    assertAtEOF(stream);
+  }
 }