[format] Close the codec probe stream in HadoopCompressionUtils

paimon-format/src/main/java/org/apache/paimon/format/text/AbstractTextFileWriter.java

@@ -59,7 +59,8 @@ protected AbstractTextFileWriter(
 
     @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();
     }
 

paimon-format/src/main/java/org/apache/paimon/format/text/HadoopCompressionUtils.java

@@ -28,6 +28,7 @@
 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 static Optional<CompressionCodec> getCompressionCodecByCompression(String
                 ((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);

paimon-format/src/test/java/org/apache/paimon/format/text/HadoopCompressionUtilsTest.java

@@ -24,6 +24,8 @@
 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();