Remove unnecessary calls to CodecPool.returnCompressor/returnDecompresso...

src/main/java/com/hadoop/compression/lzo/LzopCodec.java

@@ -67,15 +67,19 @@ public class LzopCodec extends LzoCodec {
   public CompressionOutputStream createOutputStream(OutputStream out) throws IOException {
     //get a compressor which will be returned to the pool when the output stream
     //is closed.
-    return createOutputStream(out, getCompressor());
+    Compressor compressor = getCompressor();
+    OutputStream wrapped = new WrappedOutputStream(out, compressor);
+    return createOutputStream(wrapped, compressor);
   }
 
   public CompressionOutputStream createIndexedOutputStream(OutputStream out,
                                                            DataOutputStream indexOut)
                                                            throws IOException {
     //get a compressor which will be returned to the pool when the output stream
     //is closed.
-    return createIndexedOutputStream(out, indexOut, getCompressor());
+    Compressor compressor = getCompressor();
+    OutputStream wrapped = new WrappedOutputStream(out, compressor);
+    return createIndexedOutputStream(wrapped, indexOut, compressor);
   }
 
   @Override
@@ -106,11 +110,56 @@ public CompressionInputStream createInputStream(InputStream in,
             getConf().getInt(LZO_BUFFER_SIZE_KEY, DEFAULT_LZO_BUFFER_SIZE));
   }
 
+  private static class WrappedInputStream extends InputStream {
+    private InputStream inputStream;
+    private Decompressor decompressor;
+
+    public WrappedInputStream(InputStream inputStream, Decompressor decompressor) {
+      this.inputStream = inputStream;
+      this.decompressor = decompressor;
+    }
+
+    @Override
+    public int read() throws IOException {
+      return inputStream.read();
+    }
+
+    @Override
+    public void close() throws IOException {
+      CodecPool.returnDecompressor(decompressor);
+      inputStream.close();
+    }
+  }
+
+  // Previous versions of the API accidentally added/removed compressor/decompressors from the pool
+  // when they shouldn't.  These classes are kind of a hack to maintain existing behavior,
+  // while still allowing proper resource management from outside
+  private static class WrappedOutputStream extends OutputStream {
+    OutputStream outputStream;
+    Compressor compressor;
+
+    public WrappedOutputStream(OutputStream outputStream, Compressor compressor) {
+      this.outputStream = outputStream;
+      this.compressor = compressor;
+    }
+
+    @Override
+    public void write(int b) throws IOException {
+      outputStream.write(b);
+    }
+
+    @Override
+    public void close() throws IOException {
+      CodecPool.returnCompressor(compressor);
+      outputStream.close();
+    }
+  }
+
   @Override
   public CompressionInputStream createInputStream(InputStream in) throws IOException {
-    // get a decompressor from a pool which will be returned to the pool
-    // when LzoInputStream is closed
-    return createInputStream(in, CodecPool.getDecompressor(this));
+    Decompressor decompressor = CodecPool.getDecompressor(this);
+    InputStream inputStream = new WrappedInputStream(in, decompressor);
+    return createInputStream(inputStream, decompressor);
   }
 
   @Override

src/main/java/com/hadoop/compression/lzo/LzopInputStream.java

@@ -31,7 +31,6 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.io.compress.BlockDecompressorStream;
-import org.apache.hadoop.io.compress.CodecPool;
 import org.apache.hadoop.io.compress.Decompressor;
 
 public class LzopInputStream extends BlockDecompressorStream {
@@ -346,9 +345,6 @@ public void close() throws IOException {
       // LZO requires that each file ends with 4 trailing zeroes.  If we are here,
       // the file didn't.  It's not critical, though, so log and eat it in this case.
       LOG.warn("Incorrect LZO file format: file did not end with four trailing zeroes.", e);
-    } finally{
-      //return the decompressor to the pool, the function itself handles null.
-      CodecPool.returnDecompressor(decompressor);
     }
   }
 }

src/main/java/com/hadoop/compression/lzo/LzopOutputStream.java

@@ -25,7 +25,6 @@
 import java.util.zip.Adler32;
 
 import org.apache.hadoop.io.DataOutputBuffer;
-import org.apache.hadoop.io.compress.CodecPool;
 import org.apache.hadoop.io.compress.CompressorStream;
 import org.apache.hadoop.io.compress.Compressor;
 
@@ -114,9 +113,6 @@ public void close() throws IOException {
         indexOut.close();
       }
       closed = true;
-      //return the compressor to the pool for later reuse;
-      //the returnCompressor handles nulls.
-      CodecPool.returnCompressor(compressor);
     }
   }