HADOOP-3821. Prevent SequenceFile and IFile from duplicating codecs in

CodecPool when closed more than once. Contributed by Arun Murthy.


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@688985 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -371,6 +371,13 @@ Trunk (unreleased changes)
     HADOOP-3705. Fix mapred.join parser to accept InputFormats named with
     underscore and static, inner classes. (cdouglas)
 
+Release 0.18.1 - Unreleased
+
+  BUG FIXES
+
+    HADOOP-3821. Prevent SequenceFile and IFile from duplicating codecs in
+    CodecPool when closed more than once. (Arun Murthy via cdouglas)
+
 Release 0.18.0 - 2008-08-19
 
   INCOMPATIBLE CHANGES

src/core/org/apache/hadoop/io/SequenceFile.java

@@ -944,6 +944,7 @@ public class SequenceFile {
     /** Close the file. */
     public synchronized void close() throws IOException {
       CodecPool.returnCompressor(compressor);
+      compressor = null;
       
       keySerializer.close();
       uncompressedValSerializer.close();
@@ -1569,6 +1570,8 @@ public class SequenceFile {
       CodecPool.returnDecompressor(keyDecompressor);
       CodecPool.returnDecompressor(valLenDecompressor);
       CodecPool.returnDecompressor(valDecompressor);
+      keyLenDecompressor = keyDecompressor = null;
+      valLenDecompressor = valDecompressor = null;
       
       if (keyDeserializer != null) {
     	keyDeserializer.close();

src/core/org/apache/hadoop/io/compress/CodecPool.java

@@ -59,7 +59,7 @@ public class CodecPool {
         if (codecList != null) {
           synchronized (codecList) {
             if (!codecList.isEmpty()) {
-              codec = codecList.remove(0);
+              codec = codecList.remove(codecList.size()-1);
             }
           }
         }

src/mapred/org/apache/hadoop/mapred/IFile.java

@@ -123,6 +123,7 @@ class IFile {
         compressedOut.finish();
         compressedOut.resetState();
         CodecPool.returnCompressor(compressor);
+        compressor = null;
       }
 
       // Close the stream
@@ -376,6 +377,7 @@ class IFile {
       if (decompressor != null) {
         decompressor.reset();
         CodecPool.returnDecompressor(decompressor);
+        decompressor = null;
       }
       
       // Close the underlying stream

src/test/org/apache/hadoop/io/TestSequenceFile.java

@@ -404,7 +404,61 @@ public class TestSequenceFile extends TestCase {
     }
     writer.close();
   }
+
+  public void testClose() throws IOException {
+    Configuration conf = new Configuration();
+    LocalFileSystem fs = new LocalFileSystem();
+    fs.setConf(conf);
+    fs.getRawFileSystem().setConf(conf);
+  
+    // create a sequence file 1
+    Path path1 = new Path(System.getProperty("test.build.data",".")+"/test1.seq");
+    SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path1,
+        Text.class, NullWritable.class, CompressionType.BLOCK);
+    writer.append(new Text("file1-1"), NullWritable.get());
+    writer.append(new Text("file1-2"), NullWritable.get());
+    writer.close();
   
+    Path path2 = new Path(System.getProperty("test.build.data",".")+"/test2.seq");
+    writer = SequenceFile.createWriter(fs, conf, path2, Text.class,
+        NullWritable.class, CompressionType.BLOCK);
+    writer.append(new Text("file2-1"), NullWritable.get());
+    writer.append(new Text("file2-2"), NullWritable.get());
+    writer.close();
+  
+    // Create a reader which uses 4 BuiltInZLibInflater instances
+    SequenceFile.Reader reader = new SequenceFile.Reader(fs, path1, conf);
+    // Returns the 4 BuiltInZLibInflater instances to the CodecPool
+    reader.close();
+    // The second close _could_ erroneously returns the same 
+    // 4 BuiltInZLibInflater instances to the CodecPool again
+    reader.close();
+  
+    // The first reader gets 4 BuiltInZLibInflater instances from the CodecPool
+    SequenceFile.Reader reader1 = new SequenceFile.Reader(fs, path1, conf);
+    // read first value from reader1
+    Text text = new Text();
+    reader1.next(text);
+    assertEquals("file1-1", text.toString());
+    
+    // The second reader _could_ get the same 4 BuiltInZLibInflater 
+    // instances from the CodePool as reader1
+    SequenceFile.Reader reader2 = new SequenceFile.Reader(fs, path2, conf);
+    
+    // read first value from reader2
+    reader2.next(text);
+    assertEquals("file2-1", text.toString());
+    // read second value from reader1
+    reader1.next(text);
+    assertEquals("file1-2", text.toString());
+    // read second value from reader2 (this throws an exception)
+    reader2.next(text);
+    assertEquals("file2-2", text.toString());
+  
+    assertFalse(reader1.next(text));
+    assertFalse(reader2.next(text));
+  }
+
   /** For debugging and testing. */
   public static void main(String[] args) throws Exception {
     int count = 1024 * 1024;