HDFS-630. Client can exclude specific nodes in the write pipeline.

(Nicolas Spiegelberg via dhruba)



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

CHANGES.txt

@@ -19,6 +19,9 @@ Release 0.20-append - Unreleased
     blocksBeingWritten directory. 
     (Dhruba Borthakur, Nicolas Spiegelberg, Todd Lipcon via dhruba)
 
+    HDFS-630. Client can exclude specific nodes in the write pipeline.
+    (Nicolas Spiegelberg via dhruba)
+
   IMPROVEMENTS
 
   BUG FIXES

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

@@ -84,7 +84,13 @@ public class DFSClient implements FSConstants, java.io.Closeable {
   final int writePacketSize;
   private final FileSystem.Statistics stats;
   private int maxBlockAcquireFailures;
-    
+
+  /**
+   * We assume we're talking to another CDH server, which supports
+   * HDFS-630's addBlock method. If we get a RemoteException indicating
+   * it doesn't, we'll set this false and stop trying.
+   */
+  private volatile boolean serverSupportsHdfs630 = true;
  
   public static ClientProtocol createNamenode(Configuration conf) throws IOException {
     return createNamenode(NameNode.getAddress(conf), conf);
@@ -2159,6 +2165,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
     private int packetSize = 0; // write packet size, including the header.
     private int chunksPerPacket = 0;
     private DatanodeInfo[] nodes = null; // list of targets for current block
+    private ArrayList<DatanodeInfo> excludedNodes = new ArrayList<DatanodeInfo>();
     private volatile boolean hasError = false;
     private volatile int errorIndex = 0;
     private volatile IOException lastException = null;
@@ -2863,7 +2870,9 @@ public class DFSClient implements FSConstants, java.io.Closeable {
         success = false;
                 
         long startTime = System.currentTimeMillis();
-        lb = locateFollowingBlock(startTime);
+
+        DatanodeInfo[] excluded = excludedNodes.toArray(new DatanodeInfo[0]);
+        lb = locateFollowingBlock(startTime, excluded.length > 0 ? excluded : null);
         block = lb.getBlock();
         nodes = lb.getLocations();
   
@@ -2876,6 +2885,11 @@ public class DFSClient implements FSConstants, java.io.Closeable {
           LOG.info("Abandoning block " + block);
           namenode.abandonBlock(block, src, clientName);
 
+          if (errorIndex < nodes.length) {
+            LOG.debug("Excluding datanode " + nodes[errorIndex]);
+            excludedNodes.add(nodes[errorIndex]);
+          }
+
           // Connection failed.  Let's wait a little bit and retry
           retry = true;
           try {
@@ -2973,7 +2987,8 @@ public class DFSClient implements FSConstants, java.io.Closeable {
       }
     }
   
-    private LocatedBlock locateFollowingBlock(long start
+    private LocatedBlock locateFollowingBlock(long start,
+                                              DatanodeInfo[] excludedNodes
                                               ) throws IOException {     
       int retries = conf.getInt("dfs.client.block.write.locateFollowingBlock.retries", 5);
       long sleeptime = 400;
@@ -2981,7 +2996,11 @@ public class DFSClient implements FSConstants, java.io.Closeable {
         long localstart = System.currentTimeMillis();
         while (true) {
           try {
-            return namenode.addBlock(src, clientName);
+            if (serverSupportsHdfs630) {
+              return namenode.addBlock(src, clientName, excludedNodes);
+            } else {
+              return namenode.addBlock(src, clientName);
+            }
           } catch (RemoteException e) {
             IOException ue = 
               e.unwrapRemoteException(FileNotFoundException.class,
@@ -2991,7 +3010,18 @@ public class DFSClient implements FSConstants, java.io.Closeable {
             if (ue != e) { 
               throw ue; // no need to retry these exceptions
             }
-            
+
+            if (e.getMessage().startsWith(
+                  "java.io.IOException: java.lang.NoSuchMethodException: " +
+                  "org.apache.hadoop.hdfs.protocol.ClientProtocol.addBlock(" +
+                  "java.lang.String, java.lang.String, " +
+                  "[Lorg.apache.hadoop.hdfs.protocol.DatanodeInfo;)")) {
+              // We're talking to a server that doesn't implement HDFS-630.
+              // Mark that and try again
+              serverSupportsHdfs630 = false;
+              continue;
+            }
+
             if (NotReplicatedYetException.class.getName().
                 equals(e.getClassName())) {
 

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

@@ -174,10 +174,26 @@ public interface ClientProtocol extends VersionedProtocol {
    * addBlock() allocates a new block and datanodes the block data
    * should be replicated to.
    * 
+   * @deprecated use the 3-arg form below
    * @return LocatedBlock allocated block information.
    */
   public LocatedBlock addBlock(String src, String clientName) throws IOException;
 
+  /**
+   * A client that wants to write an additional block to the 
+   * indicated filename (which must currently be open for writing)
+   * should call addBlock().  
+   *
+   * addBlock() allocates a new block and datanodes the block data
+   * should be replicated to.
+   *
+   * @param excludedNodes a list of nodes that should not be allocated
+   * 
+   * @return LocatedBlock allocated block information.
+   */
+  public LocatedBlock addBlock(String src, String clientName,
+                               DatanodeInfo[] excludedNodes) throws IOException;
+
   /**
    * The client is done writing data to the given filename, and would 
    * like to complete it.  

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

@@ -38,6 +38,7 @@ import org.apache.hadoop.metrics.util.MBeanUtil;
 import org.apache.hadoop.net.CachedDNSToSwitchMapping;
 import org.apache.hadoop.net.DNSToSwitchMapping;
 import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.net.Node;
 import org.apache.hadoop.net.ScriptBasedMapping;
 import org.apache.hadoop.hdfs.server.namenode.LeaseManager.Lease;
 import org.apache.hadoop.hdfs.server.namenode.UnderReplicatedBlocks.BlockIterator;
@@ -1240,6 +1241,15 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
     return lb;
   }
 
+  /**
+   * Stub for old callers pre-HDFS-630
+   */
+  public LocatedBlock getAdditionalBlock(String src, 
+                                         String clientName
+                                         ) throws IOException {
+    return getAdditionalBlock(src, clientName, null);
+  }
+
   /**
    * The client would like to obtain an additional block for the indicated
    * filename (which is being written-to).  Return an array that consists
@@ -1252,7 +1262,8 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
    * client to "try again later".
    */
   public LocatedBlock getAdditionalBlock(String src, 
-                                         String clientName
+                                         String clientName,
+                                         List<Node> excludedNodes
                                          ) throws IOException {
     long fileLength, blockSize;
     int replication;
@@ -1283,7 +1294,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
     // choose targets for the new block tobe allocated.
     DatanodeDescriptor targets[] = replicator.chooseTarget(replication,
                                                            clientNode,
-                                                           null,
+                                                           excludedNodes,
                                                            blockSize);
     if (targets.length < this.minReplication) {
       throw new IOException("File " + src + " could only be replicated to " +

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

@@ -47,6 +47,7 @@ import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.net.Node;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authorize.AuthorizationException;
@@ -58,7 +59,10 @@ import org.apache.hadoop.security.authorize.ServiceAuthorizationManager;
 import java.io.*;
 import java.net.*;
 import java.util.Collection;
+import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Iterator;
+import java.util.List;
 
 /**********************************************************
  * NameNode serves as both directory namespace manager and
@@ -414,12 +418,30 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
   }
 
   /**
+   * Stub for 0.20 clients that don't support HDFS-630
    */
   public LocatedBlock addBlock(String src, 
                                String clientName) throws IOException {
+    return addBlock(src, clientName, null);
+  }
+
+  public LocatedBlock addBlock(String src,
+                               String clientName,
+                               DatanodeInfo[] excludedNodes)
+    throws IOException {
+
+    List<Node> excludedNodeList = null;
+    if (excludedNodes != null) {
+      // We must copy here, since this list gets modified later on
+      // in ReplicationTargetChooser
+      excludedNodeList = new ArrayList<Node>(
+        Arrays.<Node>asList(excludedNodes));
+    }
+
     stateChangeLog.debug("*BLOCK* NameNode.addBlock: file "
                          +src+" for "+clientName);
-    LocatedBlock locatedBlock = namesystem.getAdditionalBlock(src, clientName);
+    LocatedBlock locatedBlock = namesystem.getAdditionalBlock(
+      src, clientName, excludedNodeList);
     if (locatedBlock != null)
       myMetrics.numAddBlockOps.inc();
     return locatedBlock;

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

@@ -0,0 +1,55 @@
+/**
+ * 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.hdfs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import junit.*;
+import static junit.framework.Assert.fail;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+
+/**
+ * These tests make sure that DFSClient retries fetching data from DFS
+ * properly in case of errors.
+ */
+public class TestDFSClientExcludedNodes extends junit.framework.TestCase {
+
+  public void testExcludedNodes() throws IOException {
+    Configuration conf = new Configuration();
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 3, true, null);
+    FileSystem fs = cluster.getFileSystem();
+    Path filePath = new Path("/testExcludedNodes");
+
+    // kill a datanode
+    cluster.stopDataNode(AppendTestUtil.nextInt(3));
+    OutputStream out = fs.create(filePath, true, 4096);
+    out.write(20);
+
+    try {
+      out.close();
+    } catch (Exception e) {
+      fail("DataNode failure should not result in a block abort: \n" + e.getMessage());
+    }
+  }
+  
+}

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

@@ -145,6 +145,13 @@ public class TestDFSClientRetries extends TestCase {
     public LocatedBlock addBlock(String src, String clientName)
     throws IOException
     {
+      return addBlock(src, clientName, null);
+    }
+
+
+    public LocatedBlock addBlock(String src, String clientName,
+                                 DatanodeInfo[] excludedNode)
+      throws IOException {
       num_calls++;
       if (num_calls > num_calls_allowed) { 
         throw new IOException("addBlock called more times than "
@@ -158,7 +165,6 @@ public class TestDFSClientRetries extends TestCase {
     
     
     // The following methods are stub methods that are not needed by this mock class
-
     public LocatedBlocks  getBlockLocations(String src, long offset, long length) throws IOException { return null; }
 
     public void create(String src, FsPermission masked, String clientName, boolean overwrite, short replication, long blockSize) throws IOException {}