Port from 0.20-append - HDFS-1554. New semantics for recoverLease. Contributed by Hairong Kuang.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security@1164774 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -92,6 +92,8 @@ Release 0.20.205.0 - unreleased
     HDFS-1555. Disallow pipelien recovery if a file is already being
     lease recovered. (hairong)
 
+    HDFS-1554. New semantics for recoverLease. (hairong)
+
   IMPROVEMENTS
 
     MAPREDUCE-2187. Reporter sends progress during sort/merge. (Anupam Seth via

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

@@ -554,13 +554,14 @@ public class DFSClient implements FSConstants, java.io.Closeable {
   /**
    * Recover a file's lease
    * @param src a file's path
+   * @return true if the file is already closed
    * @throws IOException
    */
-  void recoverLease(String src) throws IOException {
+  boolean recoverLease(String src) throws IOException {
     checkOpen();
     
     try {
-      namenode.recoverLease(src, clientName);
+      return namenode.recoverLease(src, clientName);
     } catch (RemoteException re) {
       throw re.unwrapRemoteException(FileNotFoundException.class,
                                      AccessControlException.class);

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

@@ -189,12 +189,14 @@ public class DistributedFileSystem extends FileSystem {
   }
 
   /** 
-   * Trigger the lease reovery of a file
+   * Start the lease recovery of a file
+   *
    * @param f a file
+   * @return true if the file is already closed
    * @throws IOException if an error occurs
    */
-  public void recoverLease(Path f) throws IOException {
-    dfs.recoverLease(getPathName(f));
+  public boolean recoverLease(Path f) throws IOException {
+    return dfs.recoverLease(getPathName(f));
   }
 
   /** This optional operation is not yet supported. */

src/hdfs/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -50,9 +50,9 @@ public interface ClientProtocol extends VersionedProtocol {
    * Compared to the previous version the following changes have been introduced:
    * (Only the latest change is reflected.
    * The log of historical changes can be retrieved from the svn).
-   * 62: Introduce a lightweight recoverLease RPC
+   * 63: recoverLease return if the file is closed or not
    */
-  public static final long versionID = 62L;
+  public static final long versionID = 63L;
   
   ///////////////////////////////////////
   // File contents
@@ -134,12 +134,14 @@ public interface ClientProtocol extends VersionedProtocol {
   public LocatedBlock append(String src, String clientName) throws IOException;
   
   /**
-   * Trigger lease recovery to happen
-   * @param src path of the file to trigger lease recovery
+   * Start lease recovery
+   * 
+   * @param src path of the file to start lease recovery
    * @param clientName name of the current client
+   * @return true if the file is already closed
    * @throws IOException
    */
-  public void recoverLease(String src, String clientName) throws IOException;
+  public boolean recoverLease(String src, String clientName) throws IOException;
 
   /**
    * Set replication for an existing file.

src/hdfs/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -1216,7 +1216,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
 
     try {
       INode myFile = dir.getFileINode(src);
-      recoverLeaseInternal(myFile, src, holder, clientMachine);
+      recoverLeaseInternal(myFile, src, holder, clientMachine, false);
 
       try {
         verifyReplication(src, replication, clientMachine);
@@ -1291,16 +1291,17 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
   }
 
   /**
-   * Trigger to recover lease;
-   * When the method returns successfully, the lease has been recovered and
-   * the file is closed.
+   * Recover lease;
+   * Immediately revoke the lease of the current lease holder and start lease
+   * recovery so that the file can be forced to be closed.
    * 
-   * @param src the path of the file to trigger release
+   * @param src the path of the file to start lease recovery
    * @param holder the lease holder's name
    * @param clientMachine the client machine's name
+   * @return true if the file is already closed
    * @throws IOException
    */
-  synchronized void recoverLease(String src, String holder, String clientMachine)
+  synchronized boolean recoverLease(String src, String holder, String clientMachine)
   throws IOException {
     if (isInSafeMode()) {
       throw new SafeModeException(
@@ -1315,15 +1316,19 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
       throw new FileNotFoundException("File not found " + src);
     }
 
+    if (!inode.isUnderConstruction()) {
+      return true;
+    }
     if (isPermissionEnabled) {
       checkPathAccess(src, FsAction.WRITE);
     }
 
-    recoverLeaseInternal(inode, src, holder, clientMachine);
+    recoverLeaseInternal(inode, src, holder, clientMachine, true);
+    return false;
   }
   
   private void recoverLeaseInternal(INode fileInode, 
-      String src, String holder, String clientMachine)
+      String src, String holder, String clientMachine, boolean force)
   throws IOException {
     if (fileInode != null && fileInode.isUnderConstruction()) {
       INodeFileUnderConstruction pendingFile = (INodeFileUnderConstruction) fileInode;
@@ -1336,7 +1341,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
       // We found the lease for this file. And surprisingly the original
       // holder is trying to recreate this file. This should never occur.
       //
-      if (lease != null) {
+      if (!force && lease != null) {
         Lease leaseFile = leaseManager.getLeaseByPath(src);
         if (leaseFile != null && leaseFile.equals(lease)) { 
           throw new AlreadyBeingCreatedException(
@@ -1355,21 +1360,29 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
                     " on client " + clientMachine + 
                     " because pendingCreates is non-null but no leases found.");
       }
-      //
-      // If the original holder has not renewed in the last SOFTLIMIT 
-      // period, then start lease recovery.
-      //
-      if (lease.expiredSoftLimit()) {
-        LOG.info("startFile: recover lease " + lease + ", src=" + src +
+      if (force) {
+        // close now: no need to wait for soft lease expiration and 
+        // close only the file src
+        LOG.info("recoverLease: recover lease " + lease + ", src=" + src +
                  " from client " + pendingFile.clientName);
-        internalReleaseLease(lease, src);
+        internalReleaseLeaseOne(lease, src);
+      } else {
+        //
+        // If the original holder has not renewed in the last SOFTLIMIT 
+        // period, then start lease recovery.
+        //
+        if (lease.expiredSoftLimit()) {
+          LOG.info("startFile: recover lease " + lease + ", src=" + src +
+              " from client " + pendingFile.clientName);
+          internalReleaseLease(lease, src);
+        }
+        throw new AlreadyBeingCreatedException(
+            "failed to create file " + src + " for " + holder +
+            " on client " + clientMachine + 
+            ", because this file is already being created by " +
+            pendingFile.getClientName() + 
+            " on " + pendingFile.getClientMachine());
       }
-      throw new AlreadyBeingCreatedException(
-                    "failed to create file " + src + " for " + holder +
-                    " on client " + clientMachine + 
-                    ", because this file is already being created by " +
-                    pendingFile.getClientName() + 
-                    " on " + pendingFile.getClientMachine());
     }
 
   }

src/hdfs/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -575,9 +575,9 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
   }
 
   /** {@inheritDoc} */
-  public void recoverLease(String src, String clientName) throws IOException {
+  public boolean recoverLease(String src, String clientName) throws IOException {
     String clientMachine = getClientMachine();
-    namesystem.recoverLease(src, clientName, clientMachine);
+    return namesystem.recoverLease(src, clientName, clientMachine);
   }
   
   /** {@inheritDoc} */

src/test/org/apache/hadoop/hdfs/TestDFSClientRetries.java

@@ -289,7 +289,7 @@ public class TestDFSClientRetries extends TestCase {
 
     public void setTimes(String src, long mtime, long atime) throws IOException {}
 
-    public void recoverLease(String src, String clientName) throws IOException {}
+    public boolean recoverLease(String src, String clientName) throws IOException {return true;}
 
     public Token<DelegationTokenIdentifier> getDelegationToken(Text renewer)
         throws IOException {

src/test/org/apache/hadoop/hdfs/TestLeaseRecovery2.java

@@ -75,7 +75,7 @@ public class TestLeaseRecovery2 extends junit.framework.TestCase {
       //create a file
       DistributedFileSystem dfs = (DistributedFileSystem)cluster.getFileSystem();
       int size = AppendTestUtil.nextInt(FILE_SIZE);
-      Path filepath = createFile(dfs, size);
+      Path filepath = createFile(dfs, size, true);
 
       // set the soft limit to be 1 second so that the
       // namenode triggers lease recovery on next attempt to write-for-open.
@@ -85,60 +85,61 @@ public class TestLeaseRecovery2 extends junit.framework.TestCase {
       verifyFile(dfs, filepath, actual, size);
 
       //test recoverLease
+      // set the soft limit to be 1 hour but recoverLease should
+      // close the file immediately
+      cluster.setLeasePeriod(hardLease, hardLease);
       size = AppendTestUtil.nextInt(FILE_SIZE);
-      filepath = createFile(dfs, size);
+      filepath = createFile(dfs, size, false);
 
-      // set the soft limit to be 1 second so that the
-      // namenode triggers lease recovery on next attempt to write-for-open.
-      cluster.setLeasePeriod(softLease, hardLease);
+      // test recoverLese from a different client
+      recoverLease(filepath, null);
+      verifyFile(dfs, filepath, actual, size);
 
-      recoverLease(filepath);
+      // test recoverlease from the same client
+      size = AppendTestUtil.nextInt(FILE_SIZE);
+      filepath = createFile(dfs, size, false);
+      
+      // create another file using the same client
+      Path filepath1 = new Path("/foo" + AppendTestUtil.nextInt());
+      FSDataOutputStream stm = dfs.create(filepath1, true,
+          bufferSize, REPLICATION_NUM, BLOCK_SIZE);
+      
+      // recover the first file
+      recoverLease(filepath, dfs);
       verifyFile(dfs, filepath, actual, size);
 
+      // continue to write to the second file
+      stm.write(buffer, 0, size);
+      stm.close();
+      verifyFile(dfs, filepath1, actual, size);
     }
     finally {
       try {
-        if (cluster != null) {cluster.shutdown();}
+        if (cluster != null) {cluster.getFileSystem().close();cluster.shutdown();}
       } catch (Exception e) {
         // ignore
       }
     }
   }
   
-  private void recoverLease(Path filepath) throws IOException, InterruptedException {
-    UserGroupInformation ugi = 
-      UserGroupInformation.createUserForTesting(fakeUsername, 
-                                                new String [] { fakeGroup});
-        
-    DistributedFileSystem dfs2 = (DistributedFileSystem)
-      DFSTestUtil.getFileSystemAs(ugi, conf);
-
-    boolean done = false;
-    while (!done) {
-      try {
-        dfs2.recoverLease(filepath);
-        done = true;
-      } catch (IOException ioe) {
-        final String message = ioe.getMessage();
-        if (message.contains(AlreadyBeingCreatedException.class.getSimpleName())) {
-          AppendTestUtil.LOG.info("GOOD! got " + message);
-        }
-        else {
-          AppendTestUtil.LOG.warn("UNEXPECTED IOException", ioe);
-        }
-      }
-
-      if (!done) {
-        AppendTestUtil.LOG.info("sleep " + 1000 + "ms");
-        try {Thread.sleep(5000);} catch (InterruptedException e) {}
-      }
+  private void recoverLease(Path filepath, DistributedFileSystem dfs2) throws Exception {
+    if (dfs2==null) {
+      UserGroupInformation ugi = 
+        UserGroupInformation.createUserForTesting(fakeUsername, 
+                                                  new String [] { fakeGroup});
+      dfs2 = (DistributedFileSystem) DFSTestUtil.getFileSystemAs(ugi, conf);
+    }
+    
+    while (!dfs2.recoverLease(filepath)) {
+      AppendTestUtil.LOG.info("sleep " + 5000 + "ms");
+      Thread.sleep(5000);
     }
   }
   
   // try to re-open the file before closing the previous handle. This
   // should fail but will trigger lease recovery.
-  private Path createFile(DistributedFileSystem dfs, int size)
-  throws IOException, InterruptedException {
+  private Path createFile(DistributedFileSystem dfs, int size,
+      boolean triggerSoftLease) throws IOException, InterruptedException {
     // create a random file name
     String filestr = "/foo" + AppendTestUtil.nextInt();
     System.out.println("filestr=" + filestr);
@@ -154,8 +155,10 @@ public class TestLeaseRecovery2 extends junit.framework.TestCase {
     // sync file
     AppendTestUtil.LOG.info("sync");
     stm.sync();
-    AppendTestUtil.LOG.info("leasechecker.interruptAndJoin()");
-    dfs.dfs.leasechecker.interruptAndJoin();
+    if (triggerSoftLease) {
+      AppendTestUtil.LOG.info("leasechecker.interruptAndJoin()");
+      dfs.dfs.leasechecker.interruptAndJoin();
+    }
     return filepath;
   }