Merge -r 740076:740077 to move the change of HADOOP-5034 from main t0 branch 0.19.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.19@740081 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -77,6 +77,9 @@ Release 0.19.1 - Unreleased
     HADOOP-4862. Minor : HADOOP-3678 did not remove all the cases of 
     spurious IOExceptions logged by DataNode. (Raghu Angadi) 
 
+    HADOOP-5034. NameNode should send both replication and deletion requests
+    to DataNode in one reply to a heartbeat. (hairong)
+
 Release 0.19.0 - 2008-11-18
 
   INCOMPATIBLE CHANGES

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

@@ -673,7 +673,7 @@ public class DataNode extends Configured
           // -- Bytes remaining
           //
           lastHeartbeat = startTime;
-          DatanodeCommand cmd = namenode.sendHeartbeat(dnRegistration,
+          DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
                                                        data.getCapacity(),
                                                        data.getDfsUsed(),
                                                        data.getRemaining(),
@@ -681,7 +681,7 @@ public class DataNode extends Configured
                                                        getXceiverCount());
           myMetrics.heartbeats.inc(now() - startTime);
           //LOG.info("Just sent heartbeat, with name " + localName);
-          if (!processCommand(cmd))
+          if (!processCommand(cmds))
             continue;
         }
             
@@ -791,6 +791,27 @@ public class DataNode extends Configured
     } // while (shouldRun)
   } // offerService
 
+  /**
+   * Process an array of datanode commands
+   * 
+   * @param cmds an array of datanode commands
+   * @return true if further processing may be required or false otherwise. 
+   */
+  private boolean processCommand(DatanodeCommand[] cmds) {
+    if (cmds != null) {
+      for (DatanodeCommand cmd : cmds) {
+        try {
+          if (processCommand(cmd) == false) {
+            return false;
+          }
+        } catch (IOException ioe) {
+          LOG.warn("Error processing datanode Command", ioe);
+        }
+      }
+    }
+    return true;
+  }
+  
     /**
      * 
      * @param cmd

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

@@ -2191,10 +2191,10 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
    * If a substantial amount of time passed since the last datanode 
    * heartbeat then request an immediate block report.  
    * 
-   * @return a datanode command 
+   * @return an array of datanode commands 
    * @throws IOException
    */
-  DatanodeCommand handleHeartbeat(DatanodeRegistration nodeReg,
+  DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
       long capacity, long dfsUsed, long remaining,
       int xceiverCount, int xmitsInProgress) throws IOException {
     DatanodeCommand cmd = null;
@@ -2204,7 +2204,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
         try {
           nodeinfo = getDatanode(nodeReg);
         } catch(UnregisteredDatanodeException e) {
-          return DatanodeCommand.REGISTER;
+          return new DatanodeCommand[]{DatanodeCommand.REGISTER};
         }
           
         // Check if this datanode should actually be shutdown instead. 
@@ -2214,7 +2214,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
         }
 
         if (nodeinfo == null || !nodeinfo.isAlive) {
-          return DatanodeCommand.REGISTER;
+          return new DatanodeCommand[]{DatanodeCommand.REGISTER};
         }
 
         updateStats(nodeinfo, false);
@@ -2222,26 +2222,35 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
         updateStats(nodeinfo, true);
         
         //check lease recovery
-        if (cmd == null) {
-          cmd = nodeinfo.getLeaseRecoveryCommand(Integer.MAX_VALUE);
+        cmd = nodeinfo.getLeaseRecoveryCommand(Integer.MAX_VALUE);
+        if (cmd != null) {
+          return new DatanodeCommand[] {cmd};
         }
+      
+        ArrayList<DatanodeCommand> cmds = new ArrayList<DatanodeCommand>(2);
         //check pending replication
-        if (cmd == null) {
-          cmd = nodeinfo.getReplicationCommand(
+        cmd = nodeinfo.getReplicationCommand(
               maxReplicationStreams - xmitsInProgress);
+        if (cmd != null) {
+          cmds.add(cmd);
         }
         //check block invalidation
-        if (cmd == null) {
-          cmd = nodeinfo.getInvalidateBlocks(blockInvalidateLimit);
+        cmd = nodeinfo.getInvalidateBlocks(blockInvalidateLimit);
+        if (cmd != null) {
+          cmds.add(cmd);
+        }
+        if (!cmds.isEmpty()) {
+          return cmds.toArray(new DatanodeCommand[cmds.size()]);
         }
       }
     }
 
     //check distributed upgrade
-    if (cmd == null) {
-      cmd = getDistributedUpgradeCommand();
+    cmd = getDistributedUpgradeCommand();
+    if (cmd != null) {
+      return new DatanodeCommand[] {cmd};
     }
-    return cmd;
+    return null;
   }
 
   private void updateStats(DatanodeDescriptor node, boolean isAdded) {

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

@@ -612,10 +612,10 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
 
   /**
    * Data node notify the name node that it is alive 
-   * Return a block-oriented command for the datanode to execute.
+   * Return an array of block-oriented commands for the datanode to execute.
    * This will be either a transfer or a delete operation.
    */
-  public DatanodeCommand sendHeartbeat(DatanodeRegistration nodeReg,
+  public DatanodeCommand[] sendHeartbeat(DatanodeRegistration nodeReg,
                                        long capacity,
                                        long dfsUsed,
                                        long remaining,

src/hdfs/org/apache/hadoop/hdfs/server/protocol/DatanodeProtocol.java

@@ -35,15 +35,10 @@ import org.apache.hadoop.ipc.VersionedProtocol;
  **********************************************************************/
 public interface DatanodeProtocol extends VersionedProtocol {
   /**
-   * 18: In sendHeartbeat, the capacity parameter reported was sum of 
-   *     the filesystem disk space of all the data directories. This is 
-   *     changed to exclude the reserved capacity defined by 
-   *     dfs.datanode.du.reserved. 
-   *
-   *     The new capacity reported is sum of the filesystem disk space of 
-   *     all the data directories minus the reserved capacity.
+   * 19: SendHeartbeat returns an array of DatanodeCommand objects
+   *     in stead of a DatanodeCommand object.
    */
-  public static final long versionID = 18L;
+  public static final long versionID = 19L;
   
   // error code
   final static int NOTIFY = 0;
@@ -77,11 +72,12 @@ public interface DatanodeProtocol extends VersionedProtocol {
   /**
    * sendHeartbeat() tells the NameNode that the DataNode is still
    * alive and well.  Includes some status info, too. 
-   * It also gives the NameNode a chance to return a "DatanodeCommand" object.
+   * It also gives the NameNode a chance to return 
+   * an array of "DatanodeCommand" objects.
    * A DatanodeCommand tells the DataNode to invalidate local block(s), 
    * or to copy them to other DataNodes, etc.
    */
-  public DatanodeCommand sendHeartbeat(DatanodeRegistration registration,
+  public DatanodeCommand[] sendHeartbeat(DatanodeRegistration registration,
                                        long capacity,
                                        long dfsUsed, long remaining,
                                        int xmitsInProgress,

src/test/org/apache/hadoop/hdfs/server/namenode/TestHeartbeatHandling.java

@@ -0,0 +1,88 @@
+package org.apache.hadoop.hdfs.server.namenode;
+
+import java.util.ArrayList;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.server.common.GenerationStamp;
+import org.apache.hadoop.hdfs.server.protocol.BlockCommand;
+import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand;
+import org.apache.hadoop.hdfs.server.protocol.DatanodeProtocol;
+import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
+
+import junit.framework.TestCase;
+
+/**
+ * Test if FSNamesystem handles heartbeat right
+ */
+public class TestHeartbeatHandling extends TestCase {
+  /**
+   * Test if {@link FSNamesystem#handleHeartbeat(DatanodeRegistration, long, long, long, int, int)}
+   * can pick up replication and/or invalidate requests and 
+   * observes the max limit
+   */
+  public void testHeartbeat() throws Exception {
+    final Configuration conf = new Configuration();
+    final MiniDFSCluster cluster = new MiniDFSCluster(conf, 1, true, null);
+    try {
+      cluster.waitActive();
+      final FSNamesystem namesystem = cluster.getNamesystem();
+      final DatanodeRegistration nodeReg = cluster.getDataNodes().get(0).dnRegistration;
+      DatanodeDescriptor dd = namesystem.getDatanode(nodeReg);
+      
+      final int REMAINING_BLOCKS = 1;
+      final int MAX_REPLICATE_LIMIT = conf.getInt("dfs.max-repl-streams", 2);
+      final int MAX_INVALIDATE_LIMIT = FSNamesystem.BLOCK_INVALIDATE_CHUNK;
+      final int MAX_INVALIDATE_BLOCKS = 2*MAX_INVALIDATE_LIMIT+REMAINING_BLOCKS;
+      final int MAX_REPLICATE_BLOCKS = 2*MAX_REPLICATE_LIMIT+REMAINING_BLOCKS;
+      final DatanodeDescriptor[] ONE_TARGET = new DatanodeDescriptor[1];
+
+      synchronized (namesystem.heartbeats) {
+      for (int i=0; i<MAX_REPLICATE_BLOCKS; i++) {
+        dd.addBlockToBeReplicated(
+            new Block(i, 0, GenerationStamp.FIRST_VALID_STAMP), ONE_TARGET);
+      }
+      DatanodeCommand[] cmds = namesystem.handleHeartbeat(
+          nodeReg, dd.getCapacity(), dd.getDfsUsed(), dd.getRemaining(), 0, 0);
+      assertEquals(1, cmds.length);
+      assertEquals(DatanodeProtocol.DNA_TRANSFER, cmds[0].getAction());
+      assertEquals(MAX_REPLICATE_LIMIT, ((BlockCommand)cmds[0]).getBlocks().length);
+      
+      ArrayList<Block> blockList = new ArrayList<Block>(MAX_INVALIDATE_BLOCKS);
+      for (int i=0; i<MAX_INVALIDATE_BLOCKS; i++) {
+        blockList.add(new Block(i, 0, GenerationStamp.FIRST_VALID_STAMP));
+      }
+      dd.addBlocksToBeInvalidated(blockList);
+           
+      cmds = namesystem.handleHeartbeat(
+          nodeReg, dd.getCapacity(), dd.getDfsUsed(), dd.getRemaining(), 0, 0);
+      assertEquals(2, cmds.length);
+      assertEquals(DatanodeProtocol.DNA_TRANSFER, cmds[0].getAction());
+      assertEquals(MAX_REPLICATE_LIMIT, ((BlockCommand)cmds[0]).getBlocks().length);
+      assertEquals(DatanodeProtocol.DNA_INVALIDATE, cmds[1].getAction());
+      assertEquals(MAX_INVALIDATE_LIMIT, ((BlockCommand)cmds[1]).getBlocks().length);
+      
+      cmds = namesystem.handleHeartbeat(
+          nodeReg, dd.getCapacity(), dd.getDfsUsed(), dd.getRemaining(), 0, 0);
+      assertEquals(2, cmds.length);
+      assertEquals(DatanodeProtocol.DNA_TRANSFER, cmds[0].getAction());
+      assertEquals(REMAINING_BLOCKS, ((BlockCommand)cmds[0]).getBlocks().length);
+      assertEquals(DatanodeProtocol.DNA_INVALIDATE, cmds[1].getAction());
+      assertEquals(MAX_INVALIDATE_LIMIT, ((BlockCommand)cmds[1]).getBlocks().length);
+      
+      cmds = namesystem.handleHeartbeat(
+          nodeReg, dd.getCapacity(), dd.getDfsUsed(), dd.getRemaining(), 0, 0);
+      assertEquals(1, cmds.length);
+      assertEquals(DatanodeProtocol.DNA_INVALIDATE, cmds[0].getAction());
+      assertEquals(REMAINING_BLOCKS, ((BlockCommand)cmds[0]).getBlocks().length);
+
+      cmds = namesystem.handleHeartbeat(
+          nodeReg, dd.getCapacity(), dd.getDfsUsed(), dd.getRemaining(), 0, 0);
+      assertEquals(null, cmds);
+      }
+    } finally {
+      cluster.shutdown();
+    }
+  }
+}