(paimon) branch master updated: [format] Close the codec probe stream in HadoopCompressionUtils (#9902)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new e085ce51e6 [format] Close the codec probe stream in 
HadoopCompressionUtils (#9902)
e085ce51e6 is described below

commit e085ce51e6206ac28e3dea45c59da4af38dd0a60
Author: jackylee <qcsd2...@gmail.com>
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();