MAPREDUCE-2631. Potential resource leaks in BinaryProtocol$TeeOutputStream.java. Contributed by Sunil G.

(cherry picked from commit 28660f51af)

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/IFileOutputStream.java

@@ -24,6 +24,7 @@ import java.io.FilterOutputStream;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.util.DataChecksum;
 /**
  * A Checksum output stream.
@@ -60,8 +61,11 @@ public class IFileOutputStream extends FilterOutputStream {
       return;
     }
     closed = true;
-    finish();
+    try {
-    out.close();
+      finish();
+    } finally {
+      IOUtils.closeStream(out);
+    }
   }
 
   /**

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/pipes/BinaryProtocol.java

@@ -36,6 +36,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
@@ -200,8 +201,8 @@ class BinaryProtocol<K1 extends WritableComparable, V1 extends Writable,
       file = new FileOutputStream(filename);
     }
     public void write(byte b[], int off, int len) throws IOException {
-      file.write(b,off,len);
+      file.write(b, off, len);
-      out.write(b,off,len);
+      out.write(b, off, len);
     }
 
     public void write(int b) throws IOException {
@@ -215,9 +216,12 @@ class BinaryProtocol<K1 extends WritableComparable, V1 extends Writable,
     }
 
     public void close() throws IOException {
-      flush();
+      try {
-      file.close();
+        flush();
-      out.close();
+      } finally {
+        IOUtils.closeStream(file);
+        IOUtils.closeStream(out);
+      }
     }
   }
 

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestIFileStreams.java

@@ -22,7 +22,13 @@ import org.apache.hadoop.fs.ChecksumException;
 import org.apache.hadoop.io.DataInputBuffer;
 import org.apache.hadoop.io.DataOutputBuffer;
 import org.junit.Test;
+import org.mockito.Mockito;
+
 import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
 import static org.junit.Assert.assertEquals;
 
 public class TestIFileStreams {
@@ -99,4 +105,18 @@ public class TestIFileStreams {
     fail("Did not detect bad data in checksum");
   }
 
+  @Test
+  public void testCloseStreamOnException() throws Exception {
+    OutputStream outputStream = Mockito.mock(OutputStream.class);
+    IFileOutputStream ifos = new IFileOutputStream(outputStream);
+    Mockito.doThrow(new IOException("Dummy Exception")).when(outputStream)
+        .flush();
+    try {
+      ifos.close();
+      fail("IOException is not thrown");
+    } catch (IOException ioe) {
+      assertEquals("Dummy Exception", ioe.getMessage());
+    }
+    Mockito.verify(outputStream).close();
+  }
 }