Merge -c 901026 to move the change made by HDFS-822 from trunk to bra…

…nch 0.21

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/branches/branch-0.21@901028 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -539,6 +539,9 @@ Trunk (unreleased changes)
     HDFS-101. DFS write pipeline: DFSClient sometimes does not detect second
     datanode failure. (hairong)
 
+    HDFS-822. Appends to already-finalized blocks can rename across volumes.
+    (hairong)
+
 Release 0.20.2 - Unreleased
 
   IMPROVEMENTS

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

@@ -1083,7 +1083,11 @@ private synchronized ReplicaBeingWritten append(FinalizedReplica replicaInfo,
 
     // construct a RBW replica with the new GS
     File blkfile = replicaInfo.getBlockFile();
-    FSVolume v = volumes.getNextVolume(estimateBlockLen);
+    FSVolume v = replicaInfo.getVolume();
+    if (v.getAvailable() < estimateBlockLen - replicaInfo.getNumBytes()) {
+      throw new DiskOutOfSpaceException("Insufficient space for appending to "
+          + replicaInfo);
+    }
     File newBlkFile = new File(v.rbwDir, replicaInfo.getBlockName());
     File oldmeta = replicaInfo.getMetaFile();
     ReplicaBeingWritten newReplicaInfo = new ReplicaBeingWritten(

src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestWriteToReplica.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.server.datanode.FSDataset.FSVolume;
+import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -147,10 +148,26 @@ private void setup(FSDataset dataSet) throws IOException {
 
   private void testAppend(FSDataset dataSet) throws IOException {
     long newGS = blocks[FINALIZED].getGenerationStamp()+1;
+    FSVolume v = dataSet.volumeMap.get(blocks[FINALIZED]).getVolume();
+    long available = v.getCapacity()-v.getDfsUsed();
+    long expectedLen = blocks[FINALIZED].getNumBytes();
+    try {
+      v.decDfsUsed(-available);
+      blocks[FINALIZED].setNumBytes(expectedLen+100);
+      dataSet.append(blocks[FINALIZED], newGS, expectedLen);
+      Assert.fail("Should not have space to append to an RWR replica" + blocks[RWR]);
+    } catch (DiskOutOfSpaceException e) {
+      Assert.assertTrue(e.getMessage().startsWith(
+          "Insufficient space for appending to "));
+    }
+    v.decDfsUsed(available);
+    blocks[FINALIZED].setNumBytes(expectedLen);
+
+    newGS = blocks[RBW].getGenerationStamp()+1;
     dataSet.append(blocks[FINALIZED], newGS, 
         blocks[FINALIZED].getNumBytes());  // successful
     blocks[FINALIZED].setGenerationStamp(newGS);
-
+        
     try {
       dataSet.append(blocks[TEMPORARY], blocks[TEMPORARY].getGenerationStamp()+1, 
           blocks[TEMPORARY].getNumBytes());