HDFS-4742. Fix appending to a renamed file with snapshot. Contributed by Jing Zhao

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-2802@1475903 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.HDFS-2802.txt

@@ -302,3 +302,6 @@ Branch-2802 Snapshot (Unreleased)
 
   HDFS-4749. Use INodeId to identify the corresponding directory node in
   FSImage saving/loading.  (Jing Zhao via szetszwo)
+
+  HDFS-4742. Fix appending to a renamed file with snapshot.  (Jing Zhao via
+  szetszwo)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeDirectory.java

@@ -199,16 +199,18 @@ private final <N extends INodeDirectory> N replaceSelf(final N newDir) {
   }
 
   /** Replace the given child with a new child. */
-  public void replaceChild(final INode oldChild, final INode newChild) {
+  public void replaceChild(INode oldChild, final INode newChild) {
     Preconditions.checkNotNull(children);
     final int i = searchChildren(newChild.getLocalNameBytes());
     Preconditions.checkState(i >= 0);
-    Preconditions.checkState(oldChild == children.get(i));
+    Preconditions.checkState(oldChild == children.get(i)
+        || oldChild == children.get(i).asReference().getReferredINode()
+            .asReference().getReferredINode());
+    oldChild = children.get(i);
     
     if (oldChild.isReference() && !newChild.isReference()) {
       // replace the referred inode, e.g., 
       // INodeFileWithSnapshot -> INodeFileUnderConstructionWithSnapshot
-      // TODO: add a unit test for rename + append
       final INode withCount = oldChild.asReference().getReferredINode();
       withCount.asReference().setReferredINode(newChild);
     } else {
@@ -219,8 +221,7 @@ public void replaceChild(final INode oldChild, final INode newChild) {
         withCount.removeReference(oldChild.asReference());
       }
       // do the replacement
-      final INode removed = children.set(i, newChild);
+      children.set(i, newChild);
-      Preconditions.checkState(removed == oldChild);
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeWithAdditionalFields.java

@@ -109,7 +109,8 @@ private INodeWithAdditionalFields(INode parent, long id, byte[] name,
   
   /** @param other Other node to be copied */
   INodeWithAdditionalFields(INodeWithAdditionalFields other) {
-    this(other.getParent(), other.getId(), other.getLocalNameBytes(),
+    this(other.getParentReference() != null ? other.getParentReference()
+        : other.getParent(), other.getId(), other.getLocalNameBytes(),
         other.permission, other.modificationTime, other.accessTime);
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java

@@ -256,7 +256,9 @@ public boolean expiredSoftLimit() {
     private String findPath(INodeFileUnderConstruction pendingFile) {
       try {
         for (String src : paths) {
-          if (fsnamesystem.dir.getINode(src) == pendingFile) {
+          INode node = fsnamesystem.dir.getINode(src);
+          if (node == pendingFile
+              || (node.isFile() && node.asFile() == pendingFile)) {
             return src;
           }
         }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestRenameWithSnapshots.java

@@ -30,10 +30,12 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.List;
+import java.util.Random;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Options.Rename;
 import org.apache.hadoop.fs.Path;
@@ -1184,6 +1186,50 @@ public void testRenameDirAndDeleteSnapshot_2() throws Exception {
     assertEquals(0, fsdir.getRoot().getDiskspace());
   }
   
+  /**
+   * Rename a file and then append the same file. 
+   */
+  @Test
+  public void testRenameAndAppend() throws Exception {
+    final Path sdir1 = new Path("/dir1");
+    final Path sdir2 = new Path("/dir2");
+    hdfs.mkdirs(sdir1);
+    hdfs.mkdirs(sdir2);
+    
+    final Path foo = new Path(sdir1, "foo");
+    DFSTestUtil.createFile(hdfs, foo, BLOCKSIZE, REPL, SEED);
+    
+    SnapshotTestHelper.createSnapshot(hdfs, sdir1, snap1);
+
+    final Path foo2 = new Path(sdir2, "foo");
+    hdfs.rename(foo, foo2);
+    
+    INode fooRef = fsdir.getINode4Write(foo2.toString());
+    assertTrue(fooRef instanceof INodeReference.DstReference);
+    
+    FSDataOutputStream out = hdfs.append(foo2);
+    try {
+      byte[] content = new byte[1024];
+      (new Random()).nextBytes(content);
+      out.write(content);
+      fooRef = fsdir.getINode4Write(foo2.toString());
+      assertTrue(fooRef instanceof INodeReference.DstReference);
+      INode fooNode = fooRef.asFile();
+      assertTrue(fooNode instanceof INodeFileUnderConstructionWithSnapshot);
+    } finally {
+      if (out != null) {
+        out.close();
+      }
+    }
+    
+    fooRef = fsdir.getINode4Write(foo2.toString());
+    assertTrue(fooRef instanceof INodeReference.DstReference);
+    INode fooNode = fooRef.asFile();
+    assertTrue(fooNode instanceof INodeFileWithSnapshot);
+    
+    restartClusterAndCheckImage();
+  }
+  
   /**
    * Test the undo section of rename. Before the rename, we create the renamed 
    * file/dir before taking the snapshot.