HDFS: Fix DFSClient's memory leak on OutputStreams hold by FileChecker

Summary:
Currently, OutputStreams are not guaranteed to be removed from DFSClient.filechecker, if OutStream.close() fails. So if clients give it up after the close() failure (which is the recommandation), the OutputStream object is leaked to FileChecker.pendingCreates forever.

This patch tries to fix the issue by always removing files from filechecker when OutputStream.close() is called. Also, the file is removed from the map when recoverLease() is called

Test Plan: ant test

Reviewers: hkuang, tomasz, weiyan

Reviewed By: hkuang

Task ID: 1862779

src/hdfs/org/apache/hadoop/hdfs/DFSClient.java

@@ -954,6 +954,9 @@ public OutputStream create(String src,
    */
   boolean recoverLease(String src, boolean discardLastBlock) throws IOException {
     checkOpen();
+    // We remove the file from local lease checker. Usually it is already been
+    // removed by the client but we want to be extra safe.
+    leasechecker.remove(src);
 
     if (this.namenodeProtocolProxy == null) {
       return versionBasedRecoverLease(src);

src/hdfs/org/apache/hadoop/hdfs/DFSOutputStream.java

@@ -1657,27 +1657,36 @@ private void waitForAckedSeqno(long seqnumToWaitFor) throws IOException {
    */
   @Override
   public void close() throws IOException {
-    if (closed) {
-      IOException e = lastException;
-      if (e == null)
-        return;
-      else
-        throw e;
-    }
-
     try {
-      closeInternal();
-      dfsClient.leasechecker.remove(src);
+      if (closed) {
+        IOException e = lastException;
+        if (e == null)
+          return;
+        else
+          throw e;
+      }
+
+      try {
+        closeInternal();
 
-      if (s != null) {
-        for (int i = 0; i < s.length; i++) {
-          s[i].close();
+        if (s != null) {
+          for (int i = 0; i < s.length; i++) {
+            s[i].close();
+          }
+          s = null;
         }
-        s = null;
+      } catch (IOException e) {
+        lastException = e;
+        throw e;
       }
-    } catch (IOException e) {
-      lastException = e;
-      throw e;
+    } finally {
+      // We always try to remove the connection from the lease to
+      // avoid memory leak. In case of failed close(), it is possible
+      // that later users' retry of close() could succeed but fail on
+      // lease expiration. Since clients don't possibly write more data
+      // after calling close(), this case doesn't change any guarantee
+      // of data itself.
+      dfsClient.leasechecker.remove(src);
     }
   }