HADOOP-2401. Only the current leaseholder can abandon a block for

a HDFS file.  ClientProtocol version changed from 20 to 21.
(Tsz Wo (Nicholas), SZE via dhruba)



git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@603421 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -29,6 +29,10 @@ Trunk (unreleased changes)
     HADOOP-2185.  RPC Server uses any available port if the specified port is
     zero. Otherwise it uses the specified port.
     (Konstantin Shvachko via dhruba)
+
+    HADOOP-2401.  Only the current leaseholder can abandon a block for
+    a HDFS file.  ClientProtocol version changed from 20 to 21.
+    (Tsz Wo (Nicholas), SZE via dhruba)
     
   NEW FEATURES
 

src/java/org/apache/hadoop/dfs/ClientProtocol.java

@@ -35,9 +35,10 @@ interface ClientProtocol extends VersionedProtocol {
    * 17 : getBlockSize replaced by getPreferredBlockSize
    * 18 : datanodereport returns dead, live or all nodes.
    * 19 : rollEditLog() returns a token to uniquely identify the editfile.
-   * 20 : getContentLength reutrns the total size in bytes of a directory subtree
+   * 20 : getContentLength returns the total size in bytes of a directory subtree
+   * 21 : add lease holder as a parameter in abandonBlock(...)
    */
-  public static final long versionID = 20L;
+  public static final long versionID = 21L;
   
   ///////////////////////////////////////
   // File contents
@@ -126,7 +127,8 @@ interface ClientProtocol extends VersionedProtocol {
    *
    * Any partial writes to the block will be garbage-collected.
    */
-  public void abandonBlock(Block b, String src) throws IOException;
+  public void abandonBlock(Block b, String src, String holder
+      ) throws IOException;
 
   /**
    * A client that wants to write an additional block to the 

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

@@ -1500,7 +1500,7 @@ class DFSClient implements FSConstants {
             Thread.sleep(6000);
           } catch (InterruptedException iex) {
           }
-          namenode.abandonBlock(block, src.toString());
+          namenode.abandonBlock(block, src, clientName);
           retry = true;
           continue;
         }
@@ -1690,7 +1690,7 @@ class DFSClient implements FSConstants {
         LOG.warn("Error closing socket.", ie2);
       }
       //XXX Why are we abondoning the block? There could be retries left.
-      namenode.abandonBlock(block, src.toString());
+      namenode.abandonBlock(block, src, clientName);
     }
 
     private void internalClose() throws IOException {

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -979,19 +979,7 @@ class FSNamesystem implements FSConstants {
         throw new SafeModeException("Cannot add block to " + src, safeMode);
       }
 
-      //
-      // make sure that we still have the lease on this file
-      //
-      INodeFile iFile = dir.getFileINode(src);
-      if (iFile == null || !iFile.isUnderConstruction()) {
-        throw new LeaseExpiredException("No lease on " + src);
-      }
-      INodeFileUnderConstruction pendingFile = (INodeFileUnderConstruction) iFile;
-      if (!pendingFile.getClientName().equals(clientName)) {
-        throw new LeaseExpiredException("Lease mismatch on " + src + " owned by "
-                                        + pendingFile.getClientName()
-                                        + " and appended by " + clientName);
-      }
+      INodeFileUnderConstruction pendingFile  = checkLease(src, clientName);
 
       //
       // If we fail this, bad things happen!
@@ -1033,21 +1021,35 @@ class FSNamesystem implements FSConstants {
   /**
    * The client would like to let go of the given block
    */
-  public synchronized boolean abandonBlock(Block b, String src) throws IOException {
+  public synchronized boolean abandonBlock(Block b, String src, String holder
+      ) throws IOException {
     //
     // Remove the block from the pending creates list
     //
     NameNode.stateChangeLog.debug("BLOCK* NameSystem.abandonBlock: "
                                   +b.getBlockName()+"of file "+src);
-    INode file = dir.getFileINode(src);
-    if (file != null) {
-      dir.removeBlock(src, file, b);
-    }
+    INode file = checkLease(src, holder);
+    dir.removeBlock(src, file, b);
     NameNode.stateChangeLog.debug("BLOCK* NameSystem.abandonBlock: "
                                     + b.getBlockName()
                                     + " is removed from pendingCreates");
     return true;
   }
+  
+  // make sure that we still have the lease on this file
+  private INodeFileUnderConstruction checkLease(String src, String holder
+      ) throws IOException {
+    INode file = dir.getFileINode(src);
+    if (file == null || !file.isUnderConstruction()) {
+      throw new LeaseExpiredException("No lease on " + src);
+    }
+    INodeFileUnderConstruction pendingFile = (INodeFileUnderConstruction)file;
+    if (!pendingFile.getClientName().equals(holder)) {
+      throw new LeaseExpiredException("Lease mismatch on " + src + " owned by "
+          + pendingFile.getClientName() + " but is accessed by " + holder);
+    }
+    return pendingFile;    
+  }
 
   /**
    * Abandon the entire file in progress

src/java/org/apache/hadoop/dfs/NameNode.java

@@ -292,10 +292,11 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
   /**
    * The client needs to give up on the block.
    */
-  public void abandonBlock(Block b, String src) throws IOException {
+  public void abandonBlock(Block b, String src, String holder
+      ) throws IOException {
     stateChangeLog.debug("*BLOCK* NameNode.abandonBlock: "
                          +b.getBlockName()+" of file "+src);
-    if (!namesystem.abandonBlock(b, src)) {
+    if (!namesystem.abandonBlock(b, src, holder)) {
       throw new IOException("Cannot abandon block during write to " + src);
     }
   }

src/test/org/apache/hadoop/dfs/TestAbandonBlock.java

@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.dfs;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.net.NetUtils;
+
+public class TestAbandonBlock extends junit.framework.TestCase {
+  public static final Log LOG = LogFactory.getLog(TestAbandonBlock.class);
+  
+  private static final Configuration CONF = new Configuration();
+  static final String FILE_NAME_PREFIX
+      = "/" + TestAbandonBlock.class.getSimpleName() + "_"; 
+
+  public void testAbondanBlock() throws IOException {
+    MiniDFSCluster cluster = new MiniDFSCluster(CONF, 2, true, null);
+    FileSystem fs = cluster.getFileSystem();
+
+    String src = FILE_NAME_PREFIX + "foo";
+    FSDataOutputStream fout = null;
+    try {
+      //start writing a a file but not close it
+      fout = fs.create(new Path(src), true, 4096, (short)1, 512L);
+      for(int i = 0; i < 1024; i++) {
+        fout.write(123);
+      }
+      fout.flush();
+  
+      //try reading the block by someone
+      DFSClient dfsclient = new DFSClient(
+          NetUtils.createSocketAddr(CONF.get("fs.default.name")), CONF);
+      LocatedBlocks blocks = dfsclient.namenode.getBlockLocations(src, 0, 1);
+      LocatedBlock b = blocks.get(0); 
+      try {
+        dfsclient.namenode.abandonBlock(b.getBlock(), src, "someone");
+        //previous line should throw an exception.
+        assertTrue(false);
+      }
+      catch(IOException ioe) {
+        LOG.info("GREAT! " + StringUtils.stringifyException(ioe));
+      }
+    }
+    finally {
+      try{fout.close();} catch(Exception e) {}
+      try{fs.close();} catch(Exception e) {}
+      try{cluster.shutdown();} catch(Exception e) {}
+    }
+  }
+}