HDFS-2991. Fix case where OP_ADD would not be logged in append(). Contributed by Todd Lipcon.

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

hdfs/CHANGES.txt

@@ -26,6 +26,9 @@ Release 0.22.1 - Unreleased
     HDFS-2877. If locking of a storage dir fails, it will remove the other
     NN's lock file on exit. (todd)
 
+    HDFS-2991. Fix case where OP_ADD would not be logged in append().
+    (Todd Lipcon via shv)
+
 Release 0.22.0 - 2011-11-29
 
   INCOMPATIBLE CHANGES

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

@@ -240,8 +240,6 @@ class FSDirectory implements Closeable {
                                    +" to the file system");
       return null;
     }
-    // add create file record to log, record new generation stamp
-    fsImage.getEditLog().logOpenFile(path, newNode);
 
     if(NameNode.stateChangeLog.isDebugEnabled()) {
       NameNode.stateChangeLog.debug("DIR* FSDirectory.addFile: "

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

@@ -218,13 +218,16 @@ public class FSEditLogLoader {
           } else {
             fsDir.updateFile(oldFile, path,  blocks, mtime, atime);
             if(opcode == Ops.OP_CLOSE) {  // OP_CLOSE
-              assert oldFile.isUnderConstruction() : 
-                "File is not under construction: " + path;
+              if (!oldFile.isUnderConstruction())
+                throw new IOException(
+                    "File is not under construction: " + path);
               fsNamesys.blockManager.completeBlock(
                   oldFile, blocks.length-1, true);
-              INodeFile newFile =
+              if (oldFile.isUnderConstruction()) {
+                INodeFile newFile =
                 ((INodeFileUnderConstruction)oldFile).convertToInodeFile();
-              fsDir.replaceNode(path, oldFile, newFile);
+                fsDir.replaceNode(path, oldFile, newFile);
+              }
             } else if(! oldFile.isUnderConstruction()) {  // OP_ADD for append
               INodeFileUnderConstruction cons = new INodeFileUnderConstruction(
                   oldFile.getLocalNameBytes(),
@@ -245,8 +248,10 @@ public class FSEditLogLoader {
             fsNamesys.leaseManager.addLease(clientName, path);
           } else {  // Ops.OP_CLOSE
             numOpClose++;
-            fsNamesys.leaseManager.removeLease(
-                ((INodeFileUnderConstruction)oldFile).getClientName(), path);
+            if (oldFile.isUnderConstruction()) {
+              fsNamesys.leaseManager.removeLease(
+                  ((INodeFileUnderConstruction)oldFile).getClientName(), path);
+            }
           }
           break;
         } 

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

@@ -1401,6 +1401,9 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
           lb.setBlockToken(blockTokenSecretManager.generateToken(lb.getBlock(), 
               EnumSet.of(BlockTokenSecretManager.AccessMode.WRITE)));
         }
+
+        // add append file record to log, record lease, etc.
+        getEditLog().logOpenFile(src, cons);
         return lb;
       } else {
        // Now we can add the name to the filesystem. This file has no
@@ -1417,6 +1420,9 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
                                 "Unable to add file to namespace.");
         }
         leaseManager.addLease(newNode.getClientName(), src);
+
+        // record file record in log, record new generation stamp
+        getEditLog().logOpenFile(src, newNode);
         if (NameNode.stateChangeLog.isDebugEnabled()) {
           NameNode.stateChangeLog.debug("DIR* NameSystem.startFile: "
                                      +"add "+src+" to namespace for "+holder);

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

@@ -0,0 +1,72 @@
+package org.apache.hadoop.hdfs;
+
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
+import org.junit.Test;
+
+public class TestFileAppendRestart {
+  private static final int BLOCK_SIZE = 4096;
+
+  private void writeAndAppend(FileSystem fs, Path p,
+      int lengthForCreate, int lengthForAppend) throws IOException {
+    // Creating a file with 4096 blockSize to write multiple blocks
+    FSDataOutputStream stream = fs.create(
+        p, true, BLOCK_SIZE, (short) 1, BLOCK_SIZE);
+    try {
+      AppendTestUtil.write(stream, 0, lengthForCreate);
+      stream.close();
+
+      stream = fs.append(p);
+      AppendTestUtil.write(stream, lengthForCreate, lengthForAppend);
+      stream.close();
+    } finally {
+      IOUtils.closeStream(stream);
+    }
+
+    int totalLength = lengthForCreate + lengthForAppend; 
+    assertEquals(totalLength, fs.getFileStatus(p).getLen());
+  }
+
+  /**
+   * Regression test for HDFS-2991. Creates and appends to files
+   * where blocks start/end on block boundaries.
+   */
+  @Test
+  public void testAppendRestart() throws Exception {
+    final Configuration conf = new HdfsConfiguration();
+    // Turn off persistent IPC, so that the DFSClient can survive NN restart
+    conf.setInt(
+        CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY,
+        0);
+    MiniDFSCluster cluster = null;
+
+    FSDataOutputStream stream = null;
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
+      FileSystem fs = cluster.getFileSystem();
+
+      Path p1 = new Path("/block-boundaries");
+      writeAndAppend(fs, p1, BLOCK_SIZE, BLOCK_SIZE);
+
+      Path p2 = new Path("/not-block-boundaries");
+      writeAndAppend(fs, p2, BLOCK_SIZE/2, BLOCK_SIZE);
+
+      cluster.restartNameNode();
+
+      AppendTestUtil.check(fs, p1, 2*BLOCK_SIZE);
+      AppendTestUtil.check(fs, p2, 3*BLOCK_SIZE/2);
+    } finally {
+      IOUtils.closeStream(stream);
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
+}