HDFS-659. If the the last block is not complete, update its length with one of its replica's length stored in datanode.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/branches/HDFS-265@820078 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -84,6 +84,9 @@ Append branch (unreleased changes)
     HDFS-550. DataNode restarts may introduce corrupt/duplicated/lost replicas
     when handling detached replicas. (hairong)
 
+    HDFS-659. If the the last block is not complete, update its length with
+    one of its replica's length stored in datanode.  (szetszwo)
+
 Trunk (unreleased changes)
 
   INCOMPATIBLE CHANGES

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

@@ -1661,9 +1661,16 @@ public class DFSClient implements FSConstants, java.io.Closeable {
         }
       }
       this.locatedBlocks = newInfo;
-      this.lastBlockBeingWrittenLength = 
-          locatedBlocks.isLastBlockComplete()? 0:
-              readBlockLength(locatedBlocks.getLastLocatedBlock()); 
+      this.lastBlockBeingWrittenLength = 0;
+      if (!locatedBlocks.isLastBlockComplete()) {
+        final LocatedBlock last = locatedBlocks.getLastLocatedBlock();
+        if (last != null) {
+          final long len = readBlockLength(last);
+          last.getBlock().setNumBytes(len);
+          this.lastBlockBeingWrittenLength = len; 
+        }
+      }
+
       this.currentNode = null;
     }
 

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

@@ -156,6 +156,7 @@ public class LocatedBlock implements Writable {
   /** {@inheritDoc} */
   public String toString() {
     return getClass().getSimpleName() + "{" + b
+        + "; getBlockSize()=" + getBlockSize()
         + "; corrupt=" + corrupt
         + "; offset=" + offset
         + "; locs=" + java.util.Arrays.asList(locs)

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

@@ -329,7 +329,7 @@ public class BlockManager {
     return machineSet;
   }
 
-  List<LocatedBlock> getBlockLocations(Block[] blocks, long offset,
+  List<LocatedBlock> getBlockLocations(BlockInfo[] blocks, long offset,
       long length, int nrBlocksToReturn) throws IOException {
     int curBlk = 0;
     long curPos = 0, blkSize = 0;
@@ -359,8 +359,14 @@ public class BlockManager {
   }
 
   /** @return a LocatedBlock for the given block */
-  LocatedBlock getBlockLocation(final Block blk, final long pos
+  LocatedBlock getBlockLocation(final BlockInfo blk, final long pos
       ) throws IOException {
+    if (!blk.isComplete()) {
+      final BlockInfoUnderConstruction uc = (BlockInfoUnderConstruction)blk;
+      final DatanodeDescriptor[] locations = uc.getExpectedLocations();
+      return namesystem.createLocatedBlock(uc, locations, pos, false);
+    }
+
     // get block locations
     final int numCorruptNodes = countNodes(blk).corruptReplicas();
     final int numCorruptReplicas = corruptReplicas.numCorruptReplicas(blk);

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

@@ -734,13 +734,8 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
       }
 
       if (!last.isComplete()) {
-        final BlockInfoUnderConstruction uc = (BlockInfoUnderConstruction)last;
-        final DatanodeDescriptor[] locations = uc.getExpectedLocations();
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("locations = " + java.util.Arrays.asList(locations));
-        }
         return new LocatedBlocks(n, inode.isUnderConstruction(), locatedblocks,
-            createLocatedBlock(uc, locations, n, false), false);
+            blockManager.getBlockLocation(last, n), false);
       }
       else {
         return new LocatedBlocks(n, inode.isUnderConstruction(), locatedblocks,

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

@@ -17,13 +17,19 @@
  */
 package org.apache.hadoop.hdfs;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
 import org.apache.commons.logging.impl.Log4JLogger;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.protocol.FSConstants;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
+import org.apache.hadoop.security.UnixUserGroupInformation;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.log4j.Level;
 import org.junit.Assert;
 import org.junit.Test;
@@ -38,56 +44,106 @@ public class TestReadWhileWriting {
   private static final String DIR = "/"
       + TestReadWhileWriting.class.getSimpleName() + "/";
   private static final int BLOCK_SIZE = 8192;
-
+  
   /** Test reading while writing. */
   @Test
   public void testReadWhileWriting() throws Exception {
-    Configuration conf = new Configuration();
+    final Configuration conf = new Configuration();
+    //enable append
+    conf.setBoolean("dfs.support.append", true);
+
     // create cluster
     final MiniDFSCluster cluster = new MiniDFSCluster(conf, 3, true, null);
     try {
+      //change the lease soft limit to 1 second.
+      final long leaseSoftLimit = 1000;
+      cluster.setLeasePeriod(leaseSoftLimit, FSConstants.LEASE_HARDLIMIT_PERIOD);
+
+      //wait for the cluster
       cluster.waitActive();
       final FileSystem fs = cluster.getFileSystem();
-
-      // write to a file but not closing it.
       final Path p = new Path(DIR, "file1");
-      final FSDataOutputStream out = fs.create(p, true,
-          fs.getConf().getInt("io.file.buffer.size", 4096),
-          (short)3, BLOCK_SIZE);
-      final int size = BLOCK_SIZE/3;
-      final byte[] buffer = AppendTestUtil.randomBytes(0, size);
-      out.write(buffer, 0, size);
-      out.flush();
-      out.sync();
-
-      // able to read?
-      Assert.assertTrue(read(fs, p, size));
-
-      out.close();
+      final int half = BLOCK_SIZE/2;
+
+      //a. On Machine M1, Create file. Write half block of data.
+      //   Invoke (DFSOutputStream).fsync() on the dfs file handle.
+      //   Do not close file yet.
+      {
+        final FSDataOutputStream out = fs.create(p, true,
+            fs.getConf().getInt("io.file.buffer.size", 4096),
+            (short)3, BLOCK_SIZE);
+        write(out, 0, half);
+
+        //hflush
+        ((DFSClient.DFSOutputStream)out.getWrappedStream()).hflush();
+      }
+
+      //b. On another machine M2, open file and verify that the half-block
+      //   of data can be read successfully.
+      checkFile(p, half, conf);
+
+      /* TODO: enable the following when append is done.
+      //c. On M1, append another half block of data.  Close file on M1.
+      {
+        //sleep to make sure the lease is expired the soft limit.
+        Thread.sleep(2*leaseSoftLimit);
+
+        FSDataOutputStream out = fs.append(p);
+        write(out, 0, half);
+        out.close();
+      }
+
+      //d. On M2, open file and read 1 block of data from it. Close file.
+      checkFile(p, 2*half, conf);
+      */
     } finally {
       cluster.shutdown();
     }
   }
 
-  /** able to read? */
-  private static boolean read(FileSystem fs, Path p, int expectedsize
-      ) throws Exception {
-    //try at most 3 minutes
-    for(int i = 0; i < 360; i++) {
-      final FSDataInputStream in = fs.open(p);
-      try {
-        final int available = in.available();
-        System.out.println(i + ") in.available()=" + available);
-        Assert.assertTrue(available >= 0);
-        Assert.assertTrue(available <= expectedsize);
-        if (available == expectedsize) {
-          return true;
-        }
-      } finally {
-        in.close();
-      }
-      Thread.sleep(500);
+  static private int userCount = 0;
+  //check the file
+  static void checkFile(Path p, int expectedsize, Configuration conf
+      ) throws IOException {
+    //open the file with another user account
+    final Configuration conf2 = new Configuration(conf);
+    final String username = UserGroupInformation.getCurrentUGI().getUserName()
+        + "_" + ++userCount;
+    UnixUserGroupInformation.saveToConf(conf2,
+        UnixUserGroupInformation.UGI_PROPERTY_NAME,
+        new UnixUserGroupInformation(username, new String[]{"supergroup"}));
+    final FileSystem fs = FileSystem.get(conf2);
+    final InputStream in = fs.open(p);
+
+    //Is the data available?
+    Assert.assertTrue(available(in, expectedsize));
+
+    //Able to read?
+    for(int i = 0; i < expectedsize; i++) {
+      Assert.assertEquals((byte)i, (byte)in.read());  
     }
-    return false;
+
+    in.close();
+  }
+
+  /** Write something to a file */
+  private static void write(OutputStream out, int offset, int length
+      ) throws IOException {
+    final byte[] bytes = new byte[length];
+    for(int i = 0; i < length; i++) {
+      bytes[i] = (byte)(offset + i);
+    }
+    out.write(bytes);
+  }
+
+  /** Is the data available? */
+  private static boolean available(InputStream in, int expectedsize
+      ) throws IOException {
+    final int available = in.available();
+    System.out.println(" in.available()=" + available);
+    Assert.assertTrue(available >= 0);
+    Assert.assertTrue(available <= expectedsize);
+    return available == expectedsize;
   }
 }
+