From dd1ab571ffe517d0c647b97d21493a00375978d2 Mon Sep 17 00:00:00 2001 From: Anas Khan <83116240+anxkhn@users.noreply.github.com> Date: Mon, 6 Jul 2026 22:27:08 +0530 Subject: [PATCH] GH-3667: Close input readers when ParquetRewriter setup fails ParquetRewriter.getFileReaders opens each input reader into a local list. If opening a later file fails, close the readers already opened before rethrowing so their input streams do not leak. Preserve close failures as suppressed exceptions on the original failure. Add a regression test that verifies the earlier input stream is closed, using AssertJ assertions. Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com> --- .../hadoop/rewrite/ParquetRewriter.java | 12 ++- .../hadoop/rewrite/ParquetRewriterTest.java | 82 +++++++++++++++++++ 2 files changed, 93 insertions(+), 1 deletion(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java index 88e41626dd..45fa7d4ede 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java @@ -87,6 +87,7 @@ import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.Type; +import org.apache.parquet.util.AutoCloseables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -336,7 +337,16 @@ private Queue getFileReaders(List inputFiles, inputFile, ParquetReadOptions.builder(conf).build()); inputFileReaders.add(reader); } catch (IOException e) { - throw new IllegalArgumentException("Failed to open input file: " + inputFile, e); + IllegalArgumentException failure = + new IllegalArgumentException("Failed to open input file: " + inputFile, e); + // Close the readers already opened so their input streams do not leak, aggregating any + // close failures as suppressed exceptions on the original failure. + try { + AutoCloseables.close(inputFileReaders); + } catch (Throwable t) { + failure.addSuppressed(t); + } + throw failure; } } return inputFileReaders; diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java index 758a2b0046..bd21d54cb3 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java @@ -46,6 +46,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiFunction; import java.util.function.Function; import java.util.stream.Collectors; @@ -88,6 +89,7 @@ import org.apache.parquet.hadoop.util.TestFileBuilder; import org.apache.parquet.internal.column.columnindex.ColumnIndex; import org.apache.parquet.internal.column.columnindex.OffsetIndex; +import org.apache.parquet.io.DelegatingSeekableInputStream; import org.apache.parquet.io.InputFile; import org.apache.parquet.io.InvalidRecordException; import org.apache.parquet.io.OutputFile; @@ -772,6 +774,86 @@ public void testMergeTwoFilesNullifyAndRenamedSameColumn(RewriterTestParams para .hasMessage("Cannot nullify and rename the same column"); } + @ParameterizedTest(name = "{0}") + @MethodSource("parameters") + public void testInputFileReadersClosedWhenLaterInputFileFailsToOpen(RewriterTestParams params) throws Exception { + initTestState(params); + MessageType schema = new MessageType("schema", new PrimitiveType(OPTIONAL, INT64, "DocId")); + EncryptionTestFile file = new TestFileBuilder(conf, schema) + .withNumRecord(numRecord) + .withCodec("UNCOMPRESSED") + .withPageSize(ParquetProperties.DEFAULT_PAGE_SIZE) + .withWriterVersion(writerVersion) + .build(); + + CloseRecordingInputFile openable = + new CloseRecordingInputFile(HadoopInputFile.fromPath(new Path(file.getFileName()), conf)); + InputFile failing = new InputFile() { + @Override + public long getLength() { + return 0; + } + + @Override + public SeekableInputStream newStream() throws IOException { + throw new IOException("simulated open failure"); + } + }; + + OutputFile output = HadoopOutputFile.fromPath(new Path(outputFile), conf); + RewriteOptions options = + new RewriteOptions.Builder(parquetConf, Arrays.asList(openable, failing), output).build(); + + assertThatThrownBy(() -> new ParquetRewriter(options)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Failed to open input file: " + failing); + assertThat(openable.isStreamClosed()).isTrue(); + } + + /** + * An {@link InputFile} that delegates to another one but records whether the {@link SeekableInputStream} it handed + * out was closed, so a test can assert that already-opened readers are released on a later failure. + */ + private static class CloseRecordingInputFile implements InputFile { + private final InputFile delegate; + private final AtomicBoolean streamClosed = new AtomicBoolean(false); + + CloseRecordingInputFile(InputFile delegate) { + this.delegate = delegate; + } + + boolean isStreamClosed() { + return streamClosed.get(); + } + + @Override + public long getLength() throws IOException { + return delegate.getLength(); + } + + @Override + public SeekableInputStream newStream() throws IOException { + SeekableInputStream stream = delegate.newStream(); + return new DelegatingSeekableInputStream(stream) { + @Override + public long getPos() throws IOException { + return stream.getPos(); + } + + @Override + public void seek(long newPos) throws IOException { + stream.seek(newPos); + } + + @Override + public void close() throws IOException { + streamClosed.set(true); + super.close(); + } + }; + } + } + public void testMergeTwoFilesWithDifferentSchemaSetup( Boolean wrongSchemaInInputFile, Map renameColumns, Map maskColumns) throws Exception {