HBASE-630 In DFSOutputStream.nextBlockOutputStream(), the client can exclude specific datanodes when locating the next block

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

CHANGES.txt

@@ -28,6 +28,10 @@ Trunk (unreleased changes)
 
     HDFS-699. Add unit tests framework (Mockito) (cos, Eli Collins)
 
+    HDFS-630  In DFSOutputStream.nextBlockOutputStream(), the client can
+              exclude specific datanodes when locating the next block
+              (Cosmin Lehene via Stack)
+
   OPTIMIZATIONS
 
   BUG FIXES

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

@@ -2585,6 +2585,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
       private DataInputStream blockReplyStream;
       private ResponseProcessor response = null;
       private volatile DatanodeInfo[] nodes = null; // list of targets for current block
+      private ArrayList<DatanodeInfo> excludedNodes = new ArrayList<DatanodeInfo>();
       volatile boolean hasError = false;
       volatile int errorIndex = -1;
       private BlockConstructionStage stage;  // block construction stage
@@ -3109,7 +3110,9 @@ public class DFSClient implements FSConstants, java.io.Closeable {
           success = false;
 
           long startTime = System.currentTimeMillis();
-          lb = locateFollowingBlock(startTime);
+          DatanodeInfo[] w = excludedNodes.toArray(
+              new DatanodeInfo[excludedNodes.size()]);
+          lb = locateFollowingBlock(startTime, w.length > 0 ? w : null);
           block = lb.getBlock();
           block.setNumBytes(0);
           accessToken = lb.getAccessToken();
@@ -3125,12 +3128,16 @@ public class DFSClient implements FSConstants, java.io.Closeable {
             namenode.abandonBlock(block, src, clientName);
             block = null;
 
+            LOG.info("Excluding datanode " + nodes[errorIndex]);
+            excludedNodes.add(nodes[errorIndex]);
+
             // Connection failed.  Let's wait a little bit and retry
             retry = true;
             try {
               if (System.currentTimeMillis() - startTime > 5000) {
                 LOG.info("Waiting to find target node: " + nodes[0].getName());
               }
+              //TODO fix this timout. Extract it o a constant, maybe make it available from conf
               Thread.sleep(6000);
             } catch (InterruptedException iex) {
             }
@@ -3228,14 +3235,15 @@ public class DFSClient implements FSConstants, java.io.Closeable {
         }
       }
 
-      private LocatedBlock locateFollowingBlock(long start) throws IOException {
+      private LocatedBlock locateFollowingBlock(long start,
+          DatanodeInfo[] excludedNodes) throws IOException {
         int retries = conf.getInt("dfs.client.block.write.locateFollowingBlock.retries", 5);
         long sleeptime = 400;
         while (true) {
           long localstart = System.currentTimeMillis();
           while (true) {
             try {
-              return namenode.addBlock(src, clientName, block);
+              return namenode.addBlock(src, clientName, block, excludedNodes);
             } catch (RemoteException e) {
               IOException ue = 
                 e.unwrapRemoteException(FileNotFoundException.class,

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

@@ -198,6 +198,9 @@ public interface ClientProtocol extends VersionedProtocol {
   public LocatedBlock addBlock(String src, String clientName,
                                Block previous) throws IOException;
 
+  public LocatedBlock addBlock(String src, String clientName,
+      Block previous, DatanodeInfo[] excludedNode) throws IOException;
+
   /**
    * The client is done writing data to the given filename, and would 
    * like to complete it.  

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

@@ -21,6 +21,7 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.net.Node;
 import org.apache.hadoop.util.ReflectionUtils;
 import java.util.*;
 
@@ -60,6 +61,26 @@ public abstract class BlockPlacementPolicy {
    * choose <i>numOfReplicas</i> data nodes for <i>writer</i> 
    * to re-replicate a block with size <i>blocksize</i> 
    * If not, return as many as we can.
+   * 
+   * @param srcPath the file to which this chooseTargets is being invoked.
+   * @param numOfReplicas additional number of replicas wanted.
+   * @param writer the writer's machine, null if not in the cluster.
+   * @param chosenNodes datanodes that have been chosen as targets.
+   * @param excludedNodes: datanodes that should not be considered as targets.
+   * @param blocksize size of the data to be written.
+   * @return array of DatanodeDescriptor instances chosen as target
+   * and sorted as a pipeline.
+   */
+  abstract DatanodeDescriptor[] chooseTarget(String srcPath,
+                                             int numOfReplicas,
+                                             DatanodeDescriptor writer,
+                                             List<DatanodeDescriptor> chosenNodes,
+                                             HashMap<Node, Node> excludedNodes,
+                                             long blocksize);
+
+  /**
+   * choose <i>numOfReplicas</i> data nodes for <i>writer</i>
+   * If not, return as many as we can.
    * The base implemenatation extracts the pathname of the file from the
    * specified srcInode, but this could be a costly operation depending on the
    * file system implementation. Concrete implementations of this class should
@@ -167,4 +188,29 @@ public abstract class BlockPlacementPolicy {
                         new ArrayList<DatanodeDescriptor>(),
                         blocksize);
   }
+
+  /**
+   * choose <i>numOfReplicas</i> nodes for <i>writer</i> to replicate
+   * a block with size <i>blocksize</i>
+   * If not, return as many as we can.
+   * 
+   * @param srcPath a string representation of the file for which chooseTarget is invoked
+   * @param numOfReplicas number of replicas wanted.
+   * @param writer the writer's machine, null if not in the cluster.
+   * @param blocksize size of the data to be written.
+   * @param excludedNodes: datanodes that should not be considered as targets.
+   * @return array of DatanodeDescriptor instances chosen as targets
+   * and sorted as a pipeline.
+   */
+  DatanodeDescriptor[] chooseTarget(String srcPath,
+                                    int numOfReplicas,
+                                    DatanodeDescriptor writer,
+                                    HashMap<Node, Node> excludedNodes,
+                                    long blocksize) {
+    return chooseTarget(srcPath, numOfReplicas, writer,
+                        new ArrayList<DatanodeDescriptor>(),
+                        excludedNodes,
+                        blocksize);
+  }
+
 }

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

@@ -67,6 +67,17 @@ public class BlockPlacementPolicyDefault extends BlockPlacementPolicy {
     return chooseTarget(numOfReplicas, writer, chosenNodes, null, blocksize);
   }
 
+  /** {@inheritDoc} */
+  public DatanodeDescriptor[] chooseTarget(String srcPath,
+                                    int numOfReplicas,
+                                    DatanodeDescriptor writer,
+                                    List<DatanodeDescriptor> chosenNodes,
+                                    HashMap<Node, Node> excludedNodes,
+                                    long blocksize) {
+    return chooseTarget(numOfReplicas, writer, chosenNodes, excludedNodes, blocksize);
+  }
+
+
   /** {@inheritDoc} */
   @Override
   public DatanodeDescriptor[] chooseTarget(FSInodeInfo srcInode,

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

@@ -40,6 +40,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.NodeBase;
 import org.apache.hadoop.net.ScriptBasedMapping;
 import org.apache.hadoop.hdfs.server.namenode.LeaseManager.Lease;
@@ -1317,10 +1318,18 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
    * are replicated.  Will return an empty 2-elt array if we want the
    * client to "try again later".
    */
-  public LocatedBlock getAdditionalBlock(String src, 
+  public LocatedBlock getAdditionalBlock(String src,
                                          String clientName,
                                          Block previous
                                          ) throws IOException {
+    return getAdditionalBlock(src, clientName, previous, null);
+  }
+
+  public LocatedBlock getAdditionalBlock(String src,
+                                         String clientName,
+                                         Block previous,
+                                         HashMap<Node, Node> excludedNodes
+                                         ) throws IOException {
     long fileLength, blockSize;
     int replication;
     DatanodeDescriptor clientNode = null;
@@ -1356,7 +1365,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
 
     // choose targets for the new block to be allocated.
     DatanodeDescriptor targets[] = blockManager.replicator.chooseTarget(
-        src, replication, clientNode, blockSize);
+        src, replication, clientNode, excludedNodes, blockSize);
     if (targets.length < blockManager.minReplication) {
       throw new IOException("File " + src + " could only be replicated to " +
                             targets.length + " nodes, instead of " +

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

@@ -22,6 +22,7 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.URI;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 
@@ -73,6 +74,7 @@ import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.Server;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.net.Node;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -612,14 +614,30 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     namesystem.setOwner(src, username, groupname);
   }
 
-  /**
-   */
+
+  @Override
   public LocatedBlock addBlock(String src, String clientName,
                                Block previous) throws IOException {
+    return addBlock(src, clientName, previous, null);
+  }
+
+  @Override
+  public LocatedBlock addBlock(String src,
+                               String clientName,
+                               Block previous,
+                               DatanodeInfo[] excludedNodes
+                               ) throws IOException {
     stateChangeLog.debug("*BLOCK* NameNode.addBlock: file "
                          +src+" for "+clientName);
+    HashMap<Node, Node> excludedNodesSet = null;
+    if (excludedNodes != null) {
+      excludedNodesSet = new HashMap<Node, Node>(excludedNodes.length);
+      for (Node node:excludedNodes) {
+        excludedNodesSet.put(node, node);
+      }
+    }
     LocatedBlock locatedBlock = 
-      namesystem.getAdditionalBlock(src, clientName, previous);
+      namesystem.getAdditionalBlock(src, clientName, previous, excludedNodesSet);
     if (locatedBlock != null)
       myMetrics.numAddBlockOps.inc();
     return locatedBlock;

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

@@ -0,0 +1,59 @@
+/**
+ * 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.framework.TestCase;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+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 TestCase {
+  public void testExcludedNodes() throws IOException
+  {   
+    Configuration conf = new HdfsConfiguration();
+    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/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java

@@ -137,8 +137,17 @@ public class TestDFSClientRetries extends TestCase {
       return versionID;
     }
 
-    public LocatedBlock addBlock(String src, String clientName, Block previous)
-    throws IOException
+    public LocatedBlock addBlock(String src, String clientName,
+                                 Block previous) throws IOException {
+
+      return addBlock(src, clientName, previous, null);
+    }
+
+    public LocatedBlock addBlock(String src,
+                                 String clientName,
+                                 Block previous,
+                                 DatanodeInfo[] excludedNode
+                                 ) throws IOException
     {
       num_calls++;
       if (num_calls > num_calls_allowed) {