HADOOP-14376. Memory leak when reading a compressed file using the native library. Contributed by Eli Acherkan

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

@@ -336,15 +336,11 @@ public class BZip2Codec implements Configurable, SplittableCompressionCodec {
     }
 
     public void close() throws IOException {
-      if (needsReset) {
+      try {
-        // In the case that nothing is written to this stream, we still need to
+        super.close();
-        // write out the header before closing, otherwise the stream won't be
+      } finally {
-        // recognized by BZip2CompressionInputStream.
+        output.close();
-        internalReset();
       }
-      this.output.flush();
-      this.output.close();
-      needsReset = true;
     }
 
   }// end of class BZip2CompressionOutputStream
@@ -454,8 +450,12 @@ public class BZip2Codec implements Configurable, SplittableCompressionCodec {
 
     public void close() throws IOException {
       if (!needsReset) {
+        try {
           input.close();
           needsReset = true;
+        } finally {
+          super.close();
+        }
       }
     }
 

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

@@ -157,7 +157,10 @@ public class CodecPool {
         LOG.debug("Got recycled compressor");
       }
     }
+    if (compressor != null &&
+        !compressor.getClass().isAnnotationPresent(DoNotPool.class)) {
       updateLeaseCount(compressorCounts, compressor, 1);
+    }
     return compressor;
   }
   
@@ -184,7 +187,10 @@ public class CodecPool {
         LOG.debug("Got recycled decompressor");
       }
     }
+    if (decompressor != null &&
+        !decompressor.getClass().isAnnotationPresent(DoNotPool.class)) {
       updateLeaseCount(decompressorCounts, decompressor, 1);
+    }
     return decompressor;
   }
   

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

@@ -59,12 +59,15 @@ public abstract class CompressionInputStream extends InputStream implements Seek
 
   @Override
   public void close() throws IOException {
+    try {
       in.close();
+    } finally {
       if (trackedDecompressor != null) {
         CodecPool.returnDecompressor(trackedDecompressor);
         trackedDecompressor = null;
       }
     }
+  }
   
   /**
    * Read bytes from the stream.

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

@@ -56,13 +56,19 @@ public abstract class CompressionOutputStream extends OutputStream {
 
   @Override
   public void close() throws IOException {
+    try {
       finish();
+    } finally {
+      try {
         out.close();
+      } finally {
         if (trackedCompressor != null) {
           CodecPool.returnCompressor(trackedCompressor);
           trackedCompressor = null;
         }
       }
+    }
+  }
   
   @Override
   public void flush() throws IOException {

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

@@ -103,10 +103,9 @@ public class CompressorStream extends CompressionOutputStream {
   public void close() throws IOException {
     if (!closed) {
       try {
-        finish();
+        super.close();
       }
       finally {
-        out.close();
         closed = true;
       }
     }

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

@@ -221,10 +221,13 @@ public class DecompressorStream extends CompressionInputStream {
   @Override
   public void close() throws IOException {
     if (!closed) {
-      in.close();
+      try {
+        super.close();
+      } finally {
         closed = true;
       }
     }
+  }
 
   @Override
   public boolean markSupported() {

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

@@ -205,28 +205,38 @@ public class TestCodec {
     
     // Compress data
     DataOutputBuffer compressedDataBuffer = new DataOutputBuffer();
-    CompressionOutputStream deflateFilter = 
+    int leasedCompressorsBefore = codec.getCompressorType() == null ? -1
+        : CodecPool.getLeasedCompressorsCount(codec);
+    try (CompressionOutputStream deflateFilter =
       codec.createOutputStream(compressedDataBuffer);
       DataOutputStream deflateOut =
-      new DataOutputStream(new BufferedOutputStream(deflateFilter));
+        new DataOutputStream(new BufferedOutputStream(deflateFilter))) {
       deflateOut.write(data.getData(), 0, data.getLength());
       deflateOut.flush();
       deflateFilter.finish();
+    }
+    if (leasedCompressorsBefore > -1) {
+      assertEquals("leased compressor not returned to the codec pool",
+          leasedCompressorsBefore, CodecPool.getLeasedCompressorsCount(codec));
+    }
     LOG.info("Finished compressing data");
     
     // De-compress data
     DataInputBuffer deCompressedDataBuffer = new DataInputBuffer();
     deCompressedDataBuffer.reset(compressedDataBuffer.getData(), 0, 
                                  compressedDataBuffer.getLength());
-    CompressionInputStream inflateFilter = 
+    DataInputBuffer originalData = new DataInputBuffer();
+    int leasedDecompressorsBefore =
+        CodecPool.getLeasedDecompressorsCount(codec);
+    try (CompressionInputStream inflateFilter =
       codec.createInputStream(deCompressedDataBuffer);
       DataInputStream inflateIn =
-      new DataInputStream(new BufferedInputStream(inflateFilter));
+        new DataInputStream(new BufferedInputStream(inflateFilter))) {
 
       // Check
-    DataInputBuffer originalData = new DataInputBuffer();
       originalData.reset(data.getData(), 0, data.getLength());
-    DataInputStream originalIn = new DataInputStream(new BufferedInputStream(originalData));
+      DataInputStream originalIn =
+          new DataInputStream(new BufferedInputStream(originalData));
       for(int i=0; i < count; ++i) {
         RandomDatum k1 = new RandomDatum();
         RandomDatum v1 = new RandomDatum();
@@ -240,7 +250,8 @@ public class TestCodec {
         assertTrue("original and compressed-then-decompressed-output not equal",
                    k1.equals(k2) && v1.equals(v2));
       
-      // original and compressed-then-decompressed-output have the same hashCode
+        // original and compressed-then-decompressed-output have the same
+        // hashCode
         Map<RandomDatum, String> m = new HashMap<RandomDatum, String>();
         m.put(k1, k1.toString());
         m.put(v1, v1.toString());
@@ -249,22 +260,28 @@ public class TestCodec {
         result = m.get(v2);
         assertEquals("v1 and v2 hashcode not equal", result, v1.toString());
       }
+    }
+    assertEquals("leased decompressor not returned to the codec pool",
+        leasedDecompressorsBefore,
+        CodecPool.getLeasedDecompressorsCount(codec));
 
     // De-compress data byte-at-a-time
     originalData.reset(data.getData(), 0, data.getLength());
     deCompressedDataBuffer.reset(compressedDataBuffer.getData(), 0, 
                                  compressedDataBuffer.getLength());
-    inflateFilter = 
+    try (CompressionInputStream inflateFilter =
       codec.createInputStream(deCompressedDataBuffer);
+      DataInputStream originalIn =
+        new DataInputStream(new BufferedInputStream(originalData))) {
 
       // Check
-    originalIn = new DataInputStream(new BufferedInputStream(originalData));
       int expected;
       do {
         expected = originalIn.read();
         assertEquals("Inflated stream read by byte does not match",
             expected, inflateFilter.read());
       } while (expected != -1);
+    }
 
     LOG.info("SUCCESS! Completed checking " + count + " records");
   }