HADOOP-18383. Codecs with @DoNotPool annotation are not closed causing memory leak

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/CodecPool.java

@@ -205,6 +205,7 @@ public static void returnCompressor(Compressor compressor) {
     }
     // if the compressor can't be reused, don't pool it.
     if (compressor.getClass().isAnnotationPresent(DoNotPool.class)) {
+      compressor.end();
       return;
     }
     compressor.reset();
@@ -225,6 +226,7 @@ public static void returnDecompressor(Decompressor decompressor) {
     }
     // if the decompressor can't be reused, don't pool it.
     if (decompressor.getClass().isAnnotationPresent(DoNotPool.class)) {
+      decompressor.end();
       return;
     }
     decompressor.reset();

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/TestCodecPool.java

@@ -18,14 +18,23 @@
 package org.apache.hadoop.io.compress;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Random;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.zlib.BuiltInGzipCompressor;
+import org.apache.hadoop.io.compress.zlib.BuiltInGzipDecompressor;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -189,4 +198,54 @@ public void testDecompressorNotReturnSameInstance() {
       CodecPool.returnDecompressor(decompressor);
     }
   }
+
+  @Test(timeout = 10000)
+  public void testDoNotPoolCompressorNotUseableAfterReturn() throws IOException {
+
+    final GzipCodec gzipCodec = new GzipCodec();
+    gzipCodec.setConf(new Configuration());
+
+    // BuiltInGzipCompressor is an explicit example of a Compressor with the @DoNotPool annotation
+    final Compressor compressor = new BuiltInGzipCompressor(new Configuration());
+    CodecPool.returnCompressor(compressor);
+
+    try (CompressionOutputStream outputStream =
+                 gzipCodec.createOutputStream(new ByteArrayOutputStream(), compressor)) {
+      outputStream.write(1);
+      fail("Compressor from Codec with @DoNotPool should not be useable after returning to CodecPool");
+    } catch (NullPointerException exception) {
+      Assert.assertEquals("Deflater has been closed", exception.getMessage());
+    }
+  }
+
+  @Test(timeout = 10000)
+  public void testDoNotPoolDecompressorNotUseableAfterReturn() throws IOException {
+
+    final GzipCodec gzipCodec = new GzipCodec();
+    gzipCodec.setConf(new Configuration());
+
+    final Random random = new Random();
+    final byte[] bytes = new byte[1024];
+    random.nextBytes(bytes);
+
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (OutputStream outputStream = gzipCodec.createOutputStream(baos)) {
+      outputStream.write(bytes);
+    }
+
+    final byte[] gzipBytes = baos.toByteArray();
+    final ByteArrayInputStream bais = new ByteArrayInputStream(gzipBytes);
+
+    // BuiltInGzipDecompressor is an explicit example of a Decompressor with the @DoNotPool annotation
+    final Decompressor decompressor = new BuiltInGzipDecompressor();
+    CodecPool.returnDecompressor(decompressor);
+
+    try (CompressionInputStream inputStream =
+                 gzipCodec.createInputStream(bais, decompressor)) {
+      inputStream.read();
+      fail("Decompressor from Codec with @DoNotPool should not be useable after returning to CodecPool");
+    } catch (NullPointerException exception) {
+      Assert.assertEquals("Inflater has been closed", exception.getMessage());
+    }
+  }
 }