This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 754bb6fe441e16bf32b488c94e8860dd44bfe43c Author: jackylee <[email protected]> AuthorDate: Thu Sep 17 13:15:48 2026 +0800 [format] Close the codec probe stream in HadoopCompressionUtils (#9902) --- .../paimon/format/text/AbstractTextFileWriter.java | 3 ++- .../paimon/format/text/HadoopCompressionUtils.java | 10 +++++++++- .../format/text/HadoopCompressionUtilsTest.java | 21 +++++++++++++++++++++ 3 files changed, 32 insertions(+), 2 deletions(-) diff --git a/paimon-format/src/main/java/org/apache/paimon/format/text/AbstractTextFileWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/text/AbstractTextFileWriter.java index 58df918507..53d8961188 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/text/AbstractTextFileWriter.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/text/AbstractTextFileWriter.java @@ -59,7 +59,8 @@ public abstract class AbstractTextFileWriter implements FormatWriter { @Override public void close() throws IOException { - writer.flush(); + // close() flushes on its way out; flushing first meant a throwing flush skipped close() + // and leaked the codec stream underneath. writer.close(); } diff --git a/paimon-format/src/main/java/org/apache/paimon/format/text/HadoopCompressionUtils.java b/paimon-format/src/main/java/org/apache/paimon/format/text/HadoopCompressionUtils.java index 6ab7734872..9d8c2642e8 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/text/HadoopCompressionUtils.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/text/HadoopCompressionUtils.java @@ -28,6 +28,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.io.compress.CompressionCodec; import org.apache.hadoop.io.compress.CompressionCodecFactory; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -114,7 +115,14 @@ public class HadoopCompressionUtils { ((Configurable) codec).setConf(new Configuration()); } - codec.createOutputStream(new java.io.ByteArrayOutputStream()); + // Opening a stream proves the codec is usable, so that a missing native library + // surfaces here rather than at write time. Hadoop leases a Compressor from CodecPool + // for the stream and takes it back only on close(), so the probe stream has to be + // closed; otherwise every call orphans one lease, and a native compressor also keeps + // its z_stream alive because that is released only through end(). + try (OutputStream ignored = codec.createOutputStream(new ByteArrayOutputStream())) { + // opening and closing is the whole probe + } return Optional.of(codec); } catch (Exception | UnsatisfiedLinkError e) { throw new RuntimeException("Failed to get compression codec", e); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/text/HadoopCompressionUtilsTest.java b/paimon-format/src/test/java/org/apache/paimon/format/text/HadoopCompressionUtilsTest.java index 0ff9ecd10e..86f09c2e8f 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/text/HadoopCompressionUtilsTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/text/HadoopCompressionUtilsTest.java @@ -24,6 +24,8 @@ import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.hadoop.io.compress.CodecPool; +import org.apache.hadoop.io.compress.CompressionCodec; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; @@ -47,6 +49,25 @@ class HadoopCompressionUtilsTest { private static final String TEST_DATA = "This is test data for compression."; + @Test + void testCodecProbeDoesNotLeaseACompressor() { + // The codec probe opens an output stream to prove the codec is usable. Hadoop leases a + // Compressor from CodecPool for that stream and takes it back only on close(), so a + // discarded probe stream orphaned one compressor per call. DEFLATE is used here because + // it needs no native library. + String deflate = HadoopCompressionType.DEFLATE.value(); + CompressionCodec codec = + HadoopCompressionUtils.getCompressionCodecByCompression(deflate) + .orElseThrow(IllegalStateException::new); + int leasedAfterFirst = CodecPool.getLeasedCompressorsCount(codec); + + for (int i = 0; i < 4; i++) { + HadoopCompressionUtils.getCompressionCodecByCompression(deflate); + } + + assertThat(CodecPool.getLeasedCompressorsCount(codec)).isEqualTo(leasedAfterFirst); + } + @Test void testCreateCompressedOutputStreamWithNoneCompression() throws IOException { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
