Revert the change made to HDFS-1555: merge -c -1056483 https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-append

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

CHANGES.txt

@@ -91,9 +91,6 @@ Release 0.20-append - Unreleased
     HDFS-724.  Use a bidirectional heartbeat to detect stuck
     pipeline. (hairong)
 
-    HDFS-1555. Disallow pipelien recovery if a file is already being
-    lease recovered. (hairong)
-
 Release 0.20.3 - Unreleased
 
   NEW FEATURES

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

@@ -180,8 +180,7 @@ public class DistributedFileSystem extends FileSystem {
   }
 
   /** 
-   * Start the lease reovery of a file
-   *
+   * Trigger the lease reovery of a file
    * @param f a file
    * @throws IOException if an error occurs
    */

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

@@ -124,9 +124,8 @@ public interface ClientProtocol extends VersionedProtocol {
   public LocatedBlock append(String src, String clientName) throws IOException;
   
   /**
-   * Start lease recovery
-   * 
-   * @param src path of the file to start lease recovery
+   * Trigger lease recovery to happen
+   * @param src path of the file to trigger lease recovery
    * @param clientName name of the current client
    * @throws IOException
    */

src/hdfs/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -1574,7 +1574,7 @@ public class DataNode extends Configured
 
     List<DatanodeID> successList = new ArrayList<DatanodeID>();
 
-    long generationstamp = namenode.nextGenerationStamp(block, closeFile);
+    long generationstamp = namenode.nextGenerationStamp(block);
     Block newblock = new Block(block.getBlockId(), block.getNumBytes(), generationstamp);
 
     for(BlockRecord r : syncList) {

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

@@ -893,23 +893,6 @@ class FSDirectory implements FSConstants, Closeable {
     return fullPathName.toString();
   }
   
-    /** Return the full path name of the specified inode */
-    static String getFullPathName(INode inode) {
-      // calculate the depth of this inode from root
-      int depth = 0;
-      for (INode i = inode; i != null; i = i.parent) {
-        depth++;
-      }
-      INode[] inodes = new INode[depth];
-      
-      // fill up the inodes in the path from this inode to root
-      for (int i = 0; i < depth; i++) {
-        inodes[depth-i-1] = inode;
-        inode = inode.parent;
-      }
-      return getFullPathName(inodes, depth-1);
-   }
-                                                          
   /**
    * Create a directory 
    * If ancestor directories do not exist, automatically create them.

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

@@ -1051,7 +1051,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
 
     try {
       INode myFile = dir.getFileINode(src);
-      recoverLeaseInternal(myFile, src, holder, clientMachine, false);
+      recoverLeaseInternal(myFile, src, holder, clientMachine);
 
       try {
         verifyReplication(src, replication, clientMachine);
@@ -1126,11 +1126,11 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
   }
 
   /**
-   * 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.
+   * Trigger to recover lease;
+   * When the method returns successfully, the lease has been recovered and
+   * the file is closed.
    * 
-   * @param src the path of the file to start lease recovery
+   * @param src the path of the file to trigger release
    * @param holder the lease holder's name
    * @param clientMachine the client machine's name
    * @throws IOException
@@ -1154,11 +1154,11 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
       checkPathAccess(src, FsAction.WRITE);
     }
 
-    recoverLeaseInternal(inode, src, holder, clientMachine, true);
+    recoverLeaseInternal(inode, src, holder, clientMachine);
   }
   
   private void recoverLeaseInternal(INode fileInode, 
-      String src, String holder, String clientMachine, boolean force)
+      String src, String holder, String clientMachine)
   throws IOException {
     if (fileInode != null && fileInode.isUnderConstruction()) {
       INodeFileUnderConstruction pendingFile = (INodeFileUnderConstruction) fileInode;
@@ -1171,7 +1171,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 (!force && lease != null) {
+      if (lease != null) {
         Lease leaseFile = leaseManager.getLeaseByPath(src);
         if (leaseFile != null && leaseFile.equals(lease)) { 
           throw new AlreadyBeingCreatedException(
@@ -1190,29 +1190,21 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
                     " on client " + clientMachine + 
                     " because pendingCreates is non-null but no leases found.");
       }
-      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 +
+      //
+      // 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);
-        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());
+        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());
     }
 
   }
@@ -4856,12 +4848,8 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
   /**
    * Verifies that the block is associated with a file that has a lease.
    * Increments, logs and then returns the stamp
-   *
-   * @param block block
-   * @param fromNN if it is for lease recovery initiated by NameNode
-   * @return a new generation stamp
-   */  
-  synchronized long nextGenerationStampForBlock(Block block, boolean fromNN) throws IOException {
+   */
+  synchronized long nextGenerationStampForBlock(Block block) throws IOException {
     if (isInSafeMode()) {
       throw new SafeModeException("Cannot get nextGenStamp for " + block, safeMode);
     }
@@ -4877,15 +4865,6 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
       LOG.info(msg);
       throw new IOException(msg);
     }
-    // Disallow client-initiated recovery once
-    // NameNode initiated lease recovery starts
-    if (!fromNN && HdfsConstants.NN_RECOVERY_LEASEHOLDER.equals(
-        leaseManager.getLeaseByPath(FSDirectory.getFullPathName(fileINode)).getHolder())) {
-      String msg = block +
-        "is being recovered by NameNode, ignoring the request from a client";
-      LOG.info(msg);
-      throw new IOException(msg);
-    }
     if (!((INodeFileUnderConstruction)fileINode).setLastRecoveryTime(now())) {
       String msg = block + " is already being recovered, ignoring this request.";
       LOG.info(msg);

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

@@ -201,12 +201,6 @@ public class LeaseManager {
       this.holder = holder;
       renew();
     }
-
-    /** Get the holder of the lease */
-    public String getHolder() {
-      return holder;
-    }
-
     /** Only LeaseManager object can renew a lease */
     private void renew() {
       this.lastUpdate = FSNamesystem.now();

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

@@ -497,8 +497,8 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
   }
 
   /** {@inheritDoc} */
-  public long nextGenerationStamp(Block block, boolean fromNN) throws IOException{
-    return namesystem.nextGenerationStampForBlock(block, fromNN);
+  public long nextGenerationStamp(Block block) throws IOException{
+    return namesystem.nextGenerationStampForBlock(block);
   }
 
   /** {@inheritDoc} */

src/hdfs/org/apache/hadoop/hdfs/server/protocol/DatanodeProtocol.java

@@ -35,10 +35,10 @@ import org.apache.hadoop.ipc.VersionedProtocol;
  **********************************************************************/
 public interface DatanodeProtocol extends VersionedProtocol {
   /**
-   * 20: nextGenerationStamp has a new parameter indicating if it is for
-   * NameNode initiated lease recovery or not
+   * 19: SendHeartbeat returns an array of DatanodeCommand objects
+   *     in stead of a DatanodeCommand object.
    */
-  public static final long versionID = 20L;
+  public static final long versionID = 19L;
   
   // error code
   final static int NOTIFY = 0;
@@ -142,14 +142,10 @@ public interface DatanodeProtocol extends VersionedProtocol {
   public void reportBadBlocks(LocatedBlock[] blocks) throws IOException;
   
   /**
-   * Get the next GenerationStamp to be associated with the specified
-   * block.
-   * 
-   * @param block block
-   * @param fromNN if it is for lease recovery initiated by NameNode
-   * @return a new generation stamp
+   * @return the next GenerationStamp to be associated with the specified
+   * block. 
    */
-  public long nextGenerationStamp(Block block, boolean fromNN) throws IOException;
+  public long nextGenerationStamp(Block block) throws IOException;
 
   /**
    * Commit block synchronization in lease recovery

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

@@ -64,7 +64,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, true);
+      Path filepath = createFile(dfs, size);
 
       // set the soft limit to be 1 second so that the
       // namenode triggers lease recovery on next attempt to write-for-open.
@@ -74,60 +74,59 @@ 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, false);
+      filepath = createFile(dfs, size);
 
-      // test recoverLese from a different client
-      recoverLease(filepath, null);
-      verifyFile(dfs, filepath, actual, size);
+      // 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 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);
+      recoverLease(filepath);
       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.getFileSystem().close(); cluster.shutdown();}
+        if (cluster != null) {cluster.shutdown();}
       } catch (Exception e) {
         // ignore
       }
     }
   }
   
-  private void recoverLease(Path filepath, DistributedFileSystem dfs2) throws Exception {
-    if (dfs2==null) {
-      Configuration conf2 = new Configuration(conf);
-      String username = UserGroupInformation.getCurrentUGI().getUserName()+"_1";
-      UnixUserGroupInformation.saveToConf(conf2,
-          UnixUserGroupInformation.UGI_PROPERTY_NAME,
-          new UnixUserGroupInformation(username, new String[]{"supergroup"}));
-      dfs2 = (DistributedFileSystem)FileSystem.get(conf2);
+  private void recoverLease(Path filepath) throws IOException {
+    Configuration conf2 = new Configuration(conf);
+    String username = UserGroupInformation.getCurrentUGI().getUserName()+"_1";
+    UnixUserGroupInformation.saveToConf(conf2,
+        UnixUserGroupInformation.UGI_PROPERTY_NAME,
+        new UnixUserGroupInformation(username, new String[]{"supergroup"}));
+    DistributedFileSystem dfs2 = (DistributedFileSystem)FileSystem.get(conf2);
+
+    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) {}
+      }
     }
-    dfs2.recoverLease(filepath);
-    checkFileClose(dfs2, filepath);
   }
   
   // 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,
-      boolean triggerSoftLease) throws IOException, InterruptedException {
+  private Path createFile(DistributedFileSystem dfs, int size) throws IOException, InterruptedException {
     // create a random file name
     String filestr = "/foo" + AppendTestUtil.nextInt();
     System.out.println("filestr=" + filestr);
@@ -143,15 +142,19 @@ public class TestLeaseRecovery2 extends junit.framework.TestCase {
     // sync file
     AppendTestUtil.LOG.info("sync");
     stm.sync();
-    if (triggerSoftLease) {
-      AppendTestUtil.LOG.info("leasechecker.interruptAndJoin()");
-      dfs.dfs.leasechecker.interruptAndJoin();
-    }
+    AppendTestUtil.LOG.info("leasechecker.interruptAndJoin()");
+    dfs.dfs.leasechecker.interruptAndJoin();
     return filepath;
   }
   
-  private void checkFileClose(FileSystem dfs2, Path filepath)
-    throws IOException {
+  private void recoverLeaseUsingCreate(Path filepath) throws IOException {
+    Configuration conf2 = new Configuration(conf);
+    String username = UserGroupInformation.getCurrentUGI().getUserName()+"_1";
+    UnixUserGroupInformation.saveToConf(conf2,
+        UnixUserGroupInformation.UGI_PROPERTY_NAME,
+        new UnixUserGroupInformation(username, new String[]{"supergroup"}));
+    FileSystem dfs2 = FileSystem.get(conf2);
+
     boolean done = false;
     for(int i = 0; i < 10 && !done; i++) {
       AppendTestUtil.LOG.info("i=" + i);
@@ -178,17 +181,7 @@ public class TestLeaseRecovery2 extends junit.framework.TestCase {
       }
     }
     assertTrue(done);
-  }
 
-  private void recoverLeaseUsingCreate(Path filepath) throws IOException {
-    Configuration conf2 = new Configuration(conf);
-    String username = UserGroupInformation.getCurrentUGI().getUserName()+"_1";
-    UnixUserGroupInformation.saveToConf(conf2,
-        UnixUserGroupInformation.UGI_PROPERTY_NAME,
-        new UnixUserGroupInformation(username, new String[]{"supergroup"}));
-    FileSystem dfs2 = FileSystem.get(conf2);
-    
-    checkFileClose(dfs2, filepath);
   }
   
   private void verifyFile(FileSystem dfs, Path filepath, byte[] actual,