HDFS-8870. Lease is leaked on write failure. Contributed by Kuhu Shukla.

(cherry picked from commit 4fcea8a0c8019d6d9a5e6f315c83659938b93a40)

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java

@@ -726,8 +726,6 @@ public class DFSOutputStream extends FSOutputSummer
         b.add(e);
       }
     }
-
-    dfsClient.endFileLease(fileId);
     final IOException ioe = b.build();
     if (ioe != null) {
       throw ioe;
@@ -740,6 +738,7 @@ public class DFSOutputStream extends FSOutputSummer
 
   void setClosed() {
     closed = true;
+    dfsClient.endFileLease(fileId);
     getStreamer().release();
   }
 
@@ -773,8 +772,6 @@ public class DFSOutputStream extends FSOutputSummer
         b.add(e);
       }
     }
-
-    dfsClient.endFileLease(fileId);
     final IOException ioe = b.build();
     if (ioe != null) {
       throw ioe;

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java

@@ -50,6 +50,11 @@ import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.anyLong;
+import org.mockito.Mockito;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import org.mockito.internal.util.reflection.Whitebox;
 
 import static org.junit.Assert.assertEquals;
@@ -202,6 +207,20 @@ public class TestDFSOutputStream {
     assertEquals(1, 3 - numDataNodesWithData);
   }
 
+  @Test
+  public void testEndLeaseCall() throws Exception {
+    Configuration conf = new Configuration();
+    DFSClient client = new DFSClient(cluster.getNameNode(0)
+        .getNameNodeAddress(), conf);
+    DFSClient spyClient = Mockito.spy(client);
+    DFSOutputStream dfsOutputStream = spyClient.create("/file2",
+        FsPermission.getFileDefault(),
+        EnumSet.of(CreateFlag.CREATE), (short) 3, 1024, null , 1024, null);
+    DFSOutputStream spyDFSOutputStream = Mockito.spy(dfsOutputStream);
+    spyDFSOutputStream.closeThreads(anyBoolean());
+    verify(spyClient, times(1)).endFileLease(anyLong());
+  }
+
   @AfterClass
   public static void tearDown() {
     if (cluster != null) {