Core: Close Avro reader when range sync fails in AvroIterable

[anxkhn]

Problem

On the split-read path, AvroIterable.iterator() opens a DataFileReader via
newFileReader() and then wraps it in an AvroRangeIterator:

FileReader<D> fileReader = initMetadata(newFileReader());   // opens the reader + stream

if (start != null) {
  ...
  fileReader = new AvroRangeIterator<>(fileReader, start, end);   // can throw
}

addCloseable(fileReader);   // ownership handed to CloseableGroup here

The AvroRangeIterator constructor calls reader.sync(start) and rethrows any
IOException as a RuntimeIOException. Because the reader is registered with the
CloseableGroup (addCloseable) only after the wrapper is built, a failure in
that sync(start) (for example a truncated or corrupt data file, or an I/O error
seeking to the split offset) propagates out of iterator() before addCloseable()
runs. The already-open DataFileReader and its underlying stream are never
registered and never closed, leaking a file handle / stream.

This is the same open-then-throw leak class that #14322 fixed for the non-split
newFileReader() path (a malformed file makes DataFileReader.openReader throw and
leak the input stream). That fix did not cover the split path, where the throw
originates in the range iterator's sync instead.

Solution

Guard the wrapper construction so the already-open reader is closed if
AvroRangeIterator throws, then rethrow:

try {
  fileReader = new AvroRangeIterator<>(fileReader, start, end);
} catch (RuntimeException e) {
  try {
    fileReader.close();
  } catch (IOException closeException) {
    e.addSuppressed(closeException);
  }
  throw e;
}

At the throw point the assignment has not completed, so fileReader still
references the base DataFileReader; close() therefore targets the correct
resource. Any secondary failure while closing is attached with addSuppressed so it
does not mask the original error. On success nothing changes: addCloseable runs as
before and the final CloseableGroup state is identical. Only the failure path now
cleans up. This mirrors the cleanup #14322 added for newFileReader().

Testing

Added a regression test readerClosedWhenRangeSyncFails in
core/src/test/java/org/apache/iceberg/avro/TestAvroIterable.java. It opens an
AvroIterable with a split (start/length set) over a mocked reader whose
sync(start) throws IOException, asserts iterator() fails with
RuntimeIOException, and verifies the reader was closed. It follows the existing
streamClosedOnIOException test in the same class (mockStatic on AvroIO and
DataFileReader).

• ./gradlew :iceberg-core:test --tests "org.apache.iceberg.avro.TestAvroIterable"
  passes (2 tests, 0 failures).
• Confirmed the test is a genuine regression: reverting only the source try/catch
  makes readerClosedWhenRangeSyncFails fail with
  Wanted but not invoked: dataFileReader.close(); restoring it turns green again.
• ./gradlew :iceberg-core:spotlessCheck passes (formatting clean).

---

AI Disclosure

• Model: Claude Opus 4.8
• Platform/Tool: opencode
• Human Oversight: fully reviewed
• Prompt Summary: Find and fix a resource leak in Iceberg core. Identified that
  AvroIterable.iterator() leaks the open DataFileReader/stream on the split-read
  path when AvroRangeIterator's sync(start) fails before the reader is registered
  with the CloseableGroup, extended the cleanup from Core: Explicitly close SeekableInput in the AvroIterable #14322 to that path, and added
  a regression test.

---

Diff summary (for reviewer reference; not part of the PR body)

• core/src/main/java/org/apache/iceberg/avro/AvroIterable.java (+10 / -1): wrap the
  AvroRangeIterator construction in try/catch (RuntimeException) that closes the
  already-open fileReader (suppressing any close IOException) and rethrows.
• core/src/test/java/org/apache/iceberg/avro/TestAvroIterable.java (+30): new
  regression test readerClosedWhenRangeSyncFails.

Total: 2 files, +40 / -1.

[wombatu-kun]

The guard starts one statement too late: setRowPositionSupplier on line 82 can throw before this try and leak the same reader. ValueReaders.PositionReader.setRowPositionSupplier calls posSupplier.get() eagerly (core/src/main/java/org/apache/iceberg/avro/ValueReaders.java:1303), so the memoized supplier runs AvroIO.findStartingRowPos right there, and that throws InvalidAvroMagicException (core/src/main/java/org/apache/iceberg/avro/AvroIO.java:163) or RuntimeIOException (AvroIO.java:181, AvroIO.java:198) on the same truncated/corrupt file and I/O-error conditions this PR targets.

It is reachable on the same split path: StructReader.setRowPositionSupplier installs a PositionReader whenever _pos is projected (ValueReaders.java:1178-1179), planned readers map _pos to ValueReaders.positions() (ValueReaders.java:320), and DeleteFilter projects MetadataColumns.ROW_POSITION whenever position deletes apply (data/src/main/java/org/apache/iceberg/data/DeleteFilter.java:325).

Move try { above the if (reader instanceof SupportsRowPosition) on line 81 so it covers both throw sites. The catch stays correct either way: fileReader is still the base reader at the first throw site, and after the assignment AvroRangeIterator.close() delegates to the same reader.

[wombatu-kun]

No test exercises this line: readerClosedWhenRangeSyncFails leaves close() unstubbed on the DataFileReader mock (core/src/test/java/org/apache/iceberg/avro/TestAvroIterable.java:76), so it returns normally and the inner catch (IOException closeException) is never entered.

Add a case that stubs doThrow(new IOException()).when(fileReader).close(); alongside the failing sync, and assert the suppressed exception on the rethrown RuntimeIOException. api/src/test/java/org/apache/iceberg/util/TestExceptionUtil.java:57-60 has the idiom: .extracting(e -> Arrays.asList(e.getSuppressed())).asInstanceOf(InstanceOfAssertFactories.LIST).hasSize(1).