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 f836feec55..ea8754aa6b 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 @@ -49,6 +49,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; @@ -91,6 +92,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; @@ -689,6 +691,84 @@ public void testMergeTwoFilesNullifyAndRenamedSameColumn() throws Exception { .hasMessage("Cannot nullify and rename the same column"); } + @Test + public void testInputFileReadersClosedWhenLaterInputFileFailsToOpen() throws Exception { + 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(); + + assertThrows(IllegalArgumentException.class, () -> new ParquetRewriter(options)); + // The reader opened for the first (valid) input file must be closed, not leaked, when a later + // input file fails to open. + assertTrue("Stream of the successfully opened input file was leaked", openable.isStreamClosed()); + } + + /** + * 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 {