HADOOP-681. Add to HDFS the ability to decommission nodes. Contributed by Dhruba.

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

CHANGES.txt

@@ -188,6 +188,10 @@ Trunk (unreleased changes)
 53. HADOOP-840.  In task tracker, queue task cleanups and perform them
     in a separate thread.  (omalley & Mahadev Konar via cutting)
 
+54. HADOOP-681.  Add to HDFS the ability to decommission nodes.  This
+    causes their blocks to be re-replicated on other nodes, so that
+    they may be removed from a cluster.  (Dhruba Borthakur via cutting)
+
 
 Release 0.9.2 - 2006-12-15
 

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

@@ -29,7 +29,7 @@ import org.apache.hadoop.ipc.VersionedProtocol;
  **********************************************************************/
 interface ClientProtocol extends VersionedProtocol {
 
-  public static final long versionID = 3L;  // setSafeMode() added
+  public static final long versionID = 4L; // decommission node added
   
     ///////////////////////////////////////
     // File contents
@@ -300,4 +300,6 @@ interface ClientProtocol extends VersionedProtocol {
      * @author Konstantin Shvachko
      */
     public boolean setSafeMode( FSConstants.SafeModeAction action ) throws IOException;
+
+    public boolean decommission( FSConstants.DecommissionAction action, String[] nodenames) throws IOException;
 }

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

@@ -86,8 +86,6 @@ public class DFSAdmin extends FsShell {
      * @exception IOException if the filesystem does not exist.
      */
     public void setSafeMode(String[] argv, int idx) throws IOException {
-      final String safeModeUsage = "Usage: java DFSAdmin -safemode "
-                                   + "[enter | leave | get]";
       if (!(fs instanceof DistributedFileSystem)) {
         System.out.println("FileSystem is " + fs.getName());
         return;
@@ -133,6 +131,65 @@ public class DFSAdmin extends FsShell {
       System.out.println("Safe mode is " + (inSafeMode ? "ON" : "OFF"));
     }
 
+    /**
+     * Command related to decommission of a datanode.
+     * Usage: java DFSAdmin -decommission [enter | leave | get]
+     * @param argv List of command line parameters. Each of these items
+              could be a hostname or a hostname:portname.
+     * @param idx The index of the command that is being processed.
+     * @exception IOException if the filesystem does not exist.
+     * @return 0 on success, non zero on error.
+     */
+    public int decommission(String[] argv, int idx) throws IOException {
+      int exitCode = -1;
+
+      if (!(fs instanceof DistributedFileSystem)) {
+        System.out.println("FileSystem is " + fs.getName());
+        return exitCode;
+      }
+      if (idx >= argv.length - 1) {
+        printUsage("-decommission");
+        return exitCode;
+      }
+      
+      //
+      // Copy all the datanode names to nodes[]
+      //
+      String[] nodes = new String[argv.length - idx - 1];
+      for (int i = idx + 1, j = 0; i < argv.length; i++, j++) {
+        nodes[j] = argv[i];
+      }
+
+      FSConstants.DecommissionAction action;
+
+      if ("set".equalsIgnoreCase(argv[idx])) {
+        action = FSConstants.DecommissionAction.DECOMMISSION_SET;
+      } else if ("clear".equalsIgnoreCase(argv[idx])) {
+        action = FSConstants.DecommissionAction.DECOMMISSION_CLEAR;
+      } else if ("get".equalsIgnoreCase(argv[idx])) {
+        action = FSConstants.DecommissionAction.DECOMMISSION_GET;
+      } else {
+        printUsage("-decommission");
+        return exitCode;
+      }
+      DistributedFileSystem dfs = (DistributedFileSystem) fs;
+      boolean mode = dfs.decommission(action, nodes);
+
+      if (action == FSConstants.DecommissionAction.DECOMMISSION_GET) {
+        if (mode) {
+          System.out.println("Node(s) has finished decommission");
+        }
+        else {
+          System.out.println("Node(s) have not yet been decommissioned");
+        }
+        return 0;
+      }
+      if (mode) {
+        return 0; // success
+      }
+      return exitCode;
+    }
+
     /**
      * Displays format of commands.
      * @param cmd The command that is being executed.
@@ -144,10 +201,15 @@ public class DFSAdmin extends FsShell {
           } else if ("-safemode".equals(cmd)) {
             System.err.println("Usage: java DFSAdmin"
                 + " [-safemode enter | leave | get | wait]");
+          } else if ("-decommission".equals(cmd)) {
+            System.err.println("Usage: java DFSAdmin"
+                + " [-decommission set | clear | get "
+                + "[datanode1[, datanode2..]]");
           } else {
             System.err.println("Usage: java DFSAdmin");
             System.err.println("           [-report]");
             System.err.println("           [-safemode enter | leave | get | wait]");
+            System.err.println("           [-decommission set | clear | get]");
           }
     }
 
@@ -180,6 +242,11 @@ public class DFSAdmin extends FsShell {
                   printUsage(cmd);
                   return exitCode;
                 }
+        } else if ("-decommission".equals(cmd)) {
+                if (argv.length < 2) {
+                  printUsage(cmd);
+                  return exitCode;
+                }
         }
 
         // initialize DFSAdmin
@@ -200,6 +267,8 @@ public class DFSAdmin extends FsShell {
                 report();
             } else if ("-safemode".equals(cmd)) {
                 setSafeMode(argv, i);
+            } else if ("-decommission".equals(cmd)) {
+                exitCode = decommission(argv, i);
             } else {
                 exitCode = -1;
                 System.err.println(cmd.substring(1) + ": Unknown command");

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

@@ -361,6 +361,18 @@ class DFSClient implements FSConstants {
       return namenode.setSafeMode( action );
     }
 
+    /**
+     * Set, clear decommission state of datnode(s).
+     * See {@link ClientProtocol#decommission(FSConstants.DecommissionAction)} 
+     * for more details.
+     * 
+     * @see ClientProtocol#decommission(FSConstants.DecommissionAction)
+     */
+    public boolean decommission(DecommissionAction action, String[] nodes)
+                                throws IOException {
+      return namenode.decommission(action, nodes);
+    }
+
     /**
      */
     public boolean mkdirs(UTF8 src) throws IOException {
@@ -528,6 +540,14 @@ class DFSClient implements FSConstants {
             this.nodes = (DatanodeInfo[][]) nodeV.toArray(new DatanodeInfo[nodeV.size()][]);
         }
 
+        /**
+         * Used by the automatic tests to detemine blocks locations of a
+         * file
+         */
+        synchronized DatanodeInfo[][] getDataNodes() {
+          return nodes;
+        }
+
         /**
          * Open a DataInputStream to a DataNode so that it can be read from.
          * We get block ID and the IDs of the destinations at startup, from the namenode.

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

@@ -27,6 +27,7 @@ import org.apache.hadoop.io.UTF8;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableFactories;
 import org.apache.hadoop.io.WritableFactory;
+import org.apache.hadoop.io.WritableUtils;
 
 /** 
  * DatanodeInfo represents the status of a DataNode.
@@ -42,8 +43,14 @@ public class DatanodeInfo extends DatanodeID {
   protected long lastUpdate;
   protected int xceiverCount;
 
+  // administrative states of a datanode
+  public enum AdminStates {NORMAL, DECOMMISSION_INPROGRESS, DECOMMISSIONED; }
+  protected AdminStates adminState;
+
+
   DatanodeInfo() {
     super();
+    adminState = null;
   }
   
   DatanodeInfo( DatanodeInfo from ) {
@@ -52,6 +59,7 @@ public class DatanodeInfo extends DatanodeID {
     this.remaining = from.getRemaining();
     this.lastUpdate = from.getLastUpdate();
     this.xceiverCount = from.getXceiverCount();
+    this.adminState = from.adminState;
   }
 
   DatanodeInfo( DatanodeID nodeID ) {
@@ -60,6 +68,7 @@ public class DatanodeInfo extends DatanodeID {
     this.remaining = 0L;
     this.lastUpdate = 0L;
     this.xceiverCount = 0;
+    this.adminState = null;
   }
   
   /** The raw capacity. */
@@ -101,6 +110,13 @@ public class DatanodeInfo extends DatanodeID {
     long r = getRemaining();
     long u = c - r;
     buffer.append("Name: "+name+"\n");
+    if (isDecommissioned()) {
+      buffer.append("State          : Decommissioned\n");
+    } else if (isDecommissionInProgress()) {
+      buffer.append("State          : Decommission in progress\n");
+    } else {
+      buffer.append("State          : In Service\n");
+    }
     buffer.append("Total raw bytes: "+c+" ("+FsShell.byteDesc(c)+")"+"\n");
     buffer.append("Used raw bytes: "+u+" ("+FsShell.byteDesc(u)+")"+"\n");
     buffer.append("% used: "+FsShell.limitDecimal(((1.0*u)/c)*100,2)+"%"+"\n");
@@ -108,6 +124,72 @@ public class DatanodeInfo extends DatanodeID {
     return buffer.toString();
   }
 
+  /**
+   * Start decommissioning a node.
+   * old state.
+   */
+  void startDecommission() {
+    adminState = AdminStates.DECOMMISSION_INPROGRESS;
+  }
+
+  /**
+   * Stop decommissioning a node.
+   * old state.
+   */
+  void stopDecommission() {
+    adminState = null;
+  }
+
+  /**
+   * Returns true if the node is in the process of being decommissioned
+   */
+   boolean isDecommissionInProgress() {
+     if (adminState == AdminStates.DECOMMISSION_INPROGRESS) {
+       return true;
+     }
+     return false;
+   }
+
+  /**
+   * Returns true if the node has been decommissioned.
+   */
+   boolean isDecommissioned() {
+     if (adminState == AdminStates.DECOMMISSIONED) {
+       return true;
+     }
+     return false;
+   }
+
+  /**
+   * Sets the admin state to indicate that decommision is complete.
+   */
+   void setDecommissioned() {
+     assert isDecommissionInProgress();
+     adminState = AdminStates.DECOMMISSIONED;
+   }
+
+   /**
+    * Retrieves the admin state of this node.
+    */
+    AdminStates getAdminState() {
+      if (adminState == null) {
+        return AdminStates.NORMAL;
+      }
+      return adminState;
+    }
+
+   /**
+    * Sets the admin state of this node.
+    */
+    void setAdminState(AdminStates newState) {
+      if (newState == AdminStates.NORMAL) {
+        adminState = null;
+      }
+      else {
+        adminState = newState;
+      }
+    }
+
   /////////////////////////////////////////////////
   // Writable
   /////////////////////////////////////////////////
@@ -127,6 +209,7 @@ public class DatanodeInfo extends DatanodeID {
     out.writeLong(remaining);
     out.writeLong(lastUpdate);
     out.writeInt(xceiverCount);
+    WritableUtils.writeEnum(out, getAdminState());
   }
 
   /**
@@ -137,5 +220,8 @@ public class DatanodeInfo extends DatanodeID {
     this.remaining = in.readLong();
     this.lastUpdate = in.readLong();
     this.xceiverCount = in.readInt();
+    AdminStates newState = (AdminStates) WritableUtils.readEnum(in,
+                                         AdminStates.class);
+    setAdminState(newState);
   }
 }

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

@@ -281,7 +281,7 @@ public class DistributedFileSystem extends FileSystem {
         return used;
     }
 
-    /** Return statistics for each datanode.*/
+    /** Return statistics for each datanode. */
     public DatanodeInfo[] getDataNodeStats() throws IOException {
       return dfs.datanodeReport();
     }
@@ -295,4 +295,13 @@ public class DistributedFileSystem extends FileSystem {
     throws IOException {
       return dfs.setSafeMode( action );
     }
+
+    /**
+     * Set, clear decommission of a set of datanodes.
+     */
+    public boolean decommission(FSConstants.DecommissionAction action,
+                                String[] nodes)
+    throws IOException {
+      return dfs.decommission(action, nodes);
+    }
 }

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

@@ -122,6 +122,9 @@ public interface FSConstants {
     // SafeMode actions
     public enum SafeModeAction{ SAFEMODE_LEAVE, SAFEMODE_ENTER, SAFEMODE_GET; }
 
+    // decommission administrative actions
+    public enum DecommissionAction{ DECOMMISSION_SET, DECOMMISSION_CLEAR, DECOMMISSION_GET; }
+
     // Version is reflected in the dfs image and edit log files.
     // Version is reflected in the data storage file.
     // Versions are negative.

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

@@ -258,8 +258,9 @@ class FSEditLog {
             if( logVersion > -3 )
               throw new IOException("Unexpected opcode " + opcode 
                   + " for version " + logVersion );
-            DatanodeDescriptor node = new DatanodeDescriptor();
-            node.readFields(in);
+            FSImage.DatanodeImage nodeimage = new FSImage.DatanodeImage();
+            nodeimage.readFields(in);
+            DatanodeDescriptor node = nodeimage.getDatanodeDescriptor();
             fsNamesys.unprotectedAddDatanode( node );
             break;
           }
@@ -376,7 +377,7 @@ class FSEditLog {
    * registration event.
    */
   void logAddDatanode( DatanodeDescriptor node ) {
-    logEdit( OP_DATANODE_ADD, node, null );
+    logEdit( OP_DATANODE_ADD, new FSImage.DatanodeImage(node), null );
   }
   
   /** 

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

@@ -74,7 +74,7 @@ class FSImage {
     }
     this.editLog = new FSEditLog( edits );
   }
-  
+
   FSEditLog getEditLog() {
     return editLog;
   }
@@ -344,7 +344,7 @@ class FSImage {
     }
   }
 
-  class DatanodeImage implements WritableComparable {
+  static class DatanodeImage implements WritableComparable {
 
     /**************************************************
      * DatanodeImage is used to store persistent information

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

@@ -717,7 +717,10 @@ class FSNamesystem implements FSConstants {
         // the blocks.
         for (int i = 0; i < nrBlocks; i++) {
           SortedSet<DatanodeDescriptor> containingNodes = blocksMap.get(pendingBlocks[i]);
-            if (containingNodes.size() < pendingFile.getReplication()) {
+          // filter out containingNodes that are marked for decommission.
+          int numCurrentReplica = countContainingNodes(containingNodes);
+
+            if (numCurrentReplica < pendingFile.getReplication()) {
                    NameNode.stateChangeLog.debug(
                           "DIR* NameSystem.completeFile:"
                         + pendingBlocks[i].getBlockName()+" has only "+containingNodes.size()
@@ -1585,20 +1588,25 @@ class FSNamesystem implements FSConstants {
             FSDirectory.INode fileINode = dir.getFileByBlock(block);
             if( fileINode == null )  // block does not belong to any file
                 return;
+
+            // filter out containingNodes that are marked for decommission.
+            int numCurrentReplica = countContainingNodes(containingNodes);
+
             // check whether safe replication is reached for the block
             // only if it is a part of a files
-            incrementSafeBlockCount( containingNodes.size() );
+            incrementSafeBlockCount( numCurrentReplica );
             short fileReplication = fileINode.getReplication();
-            if (containingNodes.size() >= fileReplication ) {
+            if (numCurrentReplica >= fileReplication ) {
                 neededReplications.remove(block);
                 pendingReplications.remove(block);
                 NameNode.stateChangeLog.trace("BLOCK* NameSystem.addStoredBlock: "
-                        +block.getBlockName()+" has "+containingNodes.size()
+                        +block.getBlockName()+" has "+ numCurrentReplica
                         +" replicas so is removed from neededReplications and pendingReplications" );
-            } else {// containingNodes.size() < fileReplication
+
+            } else {// numCurrentReplica < fileReplication
                 neededReplications.add(block);
                 NameNode.stateChangeLog.debug("BLOCK* NameSystem.addStoredBlock: "
-                    +block.getBlockName()+" has only "+containingNodes.size()
+                    +block.getBlockName()+" has only "+ numCurrentReplica
                     +" replicas so is added to neededReplications" );
             }
 
@@ -1620,7 +1628,9 @@ class FSNamesystem implements FSConstants {
           DatanodeDescriptor cur = it.next();
           Collection<Block> excessBlocks = excessReplicateMap.get(cur.getStorageID());
           if (excessBlocks == null || ! excessBlocks.contains(block)) {
+            if (!cur.isDecommissionInProgress() && !cur.isDecommissioned()) {
               nonExcess.add(cur);
+            }
           }
       }
       chooseExcessReplicates(nonExcess, block, replication);    
@@ -1811,6 +1821,145 @@ class FSNamesystem implements FSConstants {
           }
       }
     }
+
+    /**
+     * Start decommissioning the specified datanodes. If a datanode is
+     * already being decommissioned, then this is a no-op.
+     */
+    public synchronized void startDecommission (String[] nodes) 
+                             throws IOException {
+      if (isInSafeMode()) {
+        throw new SafeModeException("Cannot decommission node ", safeMode);
+      }
+      boolean isError = false;
+      String badnodes = "";
+
+      synchronized (datanodeMap) {
+        for (int i = 0; i < nodes.length; i++) {
+          boolean found = false;
+          for (Iterator<DatanodeDescriptor> it = datanodeMap.values().iterator();
+               it.hasNext(); ) {
+            DatanodeDescriptor node = it.next();
+
+            //
+            // If this is a node that we are interested in, set its admin state.
+            //
+            if (node.getName().equals(nodes[i]) || 
+                node.getHost().equals(nodes[i])) {
+              found = true;
+              if (!node.isDecommissionInProgress() && !node.isDecommissioned()) {
+                LOG.info("Start Decommissioning node " + node.name);
+                node.startDecommission();
+                //
+                // all those blocks that resides on this node has to be 
+                // replicated.
+                Block decommissionBlocks[] = node.getBlocks();
+                for (int j = 0; j < decommissionBlocks.length; j++) {
+                  synchronized (neededReplications) {
+                    neededReplications.add(decommissionBlocks[j]);
+                  }
+                }
+              }
+              break;
+            }
+          }
+          //
+          // Record the fact that a specified node was not found
+          //
+          if (!found) {
+            badnodes += nodes[i] + " ";
+            isError = true;
+          }
+        }
+      }
+      if (isError) {
+        throw new IOException("Nodes " + badnodes + " not found");
+      }
+    }
+
+    /**
+     * Stop decommissioning the specified datanodes.
+     */
+    public synchronized void stopDecommission (String[] nodes) 
+                             throws IOException {
+      if (isInSafeMode()) {
+        throw new SafeModeException("Cannot decommission node ", safeMode);
+      }
+      boolean isError = false;
+      String badnodes = "";
+
+      synchronized (datanodeMap) {
+        for (int i = 0; i < nodes.length; i++) {
+          boolean found = false;
+          for (Iterator<DatanodeDescriptor> it = datanodeMap.values().iterator();
+               it.hasNext(); ) {
+            DatanodeDescriptor node = it.next();
+
+            //
+            // If this is a node that we are interested in, set its admin state.
+            //
+            if (node.getName().equals(nodes[i]) || 
+                node.getHost().equals(nodes[i])) {
+              LOG.info("Stop Decommissioning node " + node.name);
+              found = true;
+              node.stopDecommission();
+              break;
+            }
+          }
+          //
+          // Record the fact that a specified node was not found
+          //
+          if (!found) {
+            badnodes += nodes[i] + " ";
+            isError = true;
+          }
+        }
+      }
+      if (isError) {
+        throw new IOException("Nodes " + badnodes + " not found");
+      }
+    }
+
+    /**
+     * Return true if all specified nodes are decommissioned.
+     * Otherwise return false.
+     */
+    public synchronized boolean checkDecommissioned (String[] nodes) 
+                                   throws IOException {
+      String badnodes = "";
+      boolean isError = false;
+
+      synchronized (datanodeMap) {
+        for (int i = 0; i < nodes.length; i++) {
+          boolean found = false;
+          for (Iterator<DatanodeDescriptor> it = datanodeMap.values().iterator();
+               it.hasNext(); ) {
+            DatanodeDescriptor node = it.next();
+
+            //
+            // If this is a node that we are interested in, check its admin state.
+            //
+            if (node.getName().equals(nodes[i]) || 
+                node.getHost().equals(nodes[i])) {
+              found = true;
+              boolean isDecommissioned = checkDecommissionStateInternal(node);
+              if (!isDecommissioned) {
+                return false;
+              }
+            }
+          }
+          if (!found) {
+            badnodes += nodes[i] + " ";
+            isError = true;
+          }
+        }
+      }
+      if (isError) {
+        throw new IOException("Nodes " + badnodes + " not found");
+      }
+      return true;
+    }
+
     /** 
      */
     public DatanodeInfo getDataNodeInfo(String name) {
@@ -1896,6 +2045,72 @@ class FSNamesystem implements FSConstants {
         return (Block[]) sendBlock.toArray(new Block[sendBlock.size()]);
     }
 
+    /*
+     * Counts the number of nodes in the given list. Skips over nodes
+     * that are marked for decommission.
+     */
+    private int countContainingNodes(Collection<DatanodeDescriptor> nodelist) {
+      int count = 0;
+      for (Iterator<DatanodeDescriptor> it = nodelist.iterator(); 
+           it.hasNext(); ) {
+        DatanodeDescriptor node = it.next();
+        if (!node.isDecommissionInProgress() && !node.isDecommissioned()) {
+          count++;
+        }
+      }
+      return count;
+    }
+
+    /*
+     * Return true if there are any blocks in neededReplication that 
+     * reside on the specified node. Otherwise returns false.
+     */
+    private boolean isReplicationInProgress(DatanodeDescriptor srcNode) {
+      synchronized (neededReplications) {
+        for (Iterator<Block> it = neededReplications.iterator(); it.hasNext();){
+          Block block = it.next();
+          Collection<DatanodeDescriptor> containingNodes = blocksMap.get(block); 
+          if (containingNodes.contains(srcNode)) {
+              return true;
+          }
+        }
+      }
+      return false;
+    }
+
+    /**
+     * Change, if appropriate, the admin state of a datanode to 
+     * decommission completed. Return true if decommission is complete.
+     */
+    private boolean checkDecommissionStateInternal(DatanodeDescriptor node) {
+      //
+      // Check to see if there are any blocks in the neededReplication
+      // data structure that has a replica on the node being decommissioned.
+      //
+      if (node.isDecommissionInProgress()) {
+        if (!isReplicationInProgress(node)) {
+          node.setDecommissioned();
+          LOG.info("Decommission complete for node " + node.name);
+        }
+      }
+      if (node.isDecommissioned()) {
+        return true;
+      }
+      return false;
+    }
+
+    /**
+     * Change, if appropriate, the admin state of a datanode to 
+     * decommission completed.
+     */
+    public synchronized void checkDecommissionState(DatanodeID nodeReg) {
+      DatanodeDescriptor node = datanodeMap.get(nodeReg.getStorageID());
+      if (node == null) {
+        return;
+      }
+      checkDecommissionStateInternal(node);
+    }
+
     /**
      * Return with a list of Block/DataNodeInfo sets, indicating
      * where various Blocks should be copied, ASAP.
@@ -1924,6 +2139,7 @@ class FSNamesystem implements FSConstants {
         // replicate them.
         //
         List<Block> replicateBlocks = new ArrayList<Block>();
+        List<Integer> numCurrentReplicas = new ArrayList<Integer>();
         List<DatanodeDescriptor[]> replicateTargetSets;
         replicateTargetSets = new ArrayList<DatanodeDescriptor[]>();
         for (Iterator<Block> it = neededReplications.iterator(); it.hasNext();) {
@@ -1943,17 +2159,23 @@ class FSNamesystem implements FSConstants {
             Collection<DatanodeDescriptor> containingNodes = blocksMap.get(block);
             Collection<Block> excessBlocks = excessReplicateMap.get( 
                                                       srcNode.getStorageID() );
+
             // srcNode must contain the block, and the block must
             // not be scheduled for removal on that node
             if (containingNodes != null && containingNodes.contains(srcNode)
                 && (excessBlocks == null || ! excessBlocks.contains(block))) {
+
+              // filter out containingNodes that are marked for decommission.
+              int numCurrentReplica = countContainingNodes(containingNodes);
+
               DatanodeDescriptor targets[] = chooseTargets(
-                  Math.min( fileINode.getReplication() - containingNodes.size(),
+                  Math.min( fileINode.getReplication() - numCurrentReplica,
                             this.maxReplicationStreams - xmitsInProgress), 
                   containingNodes, null, blockSize);
               if (targets.length > 0) {
                 // Build items to return
                 replicateBlocks.add(block);
+                numCurrentReplicas.add(new Integer(numCurrentReplica));
                 replicateTargetSets.add(targets);
                 scheduledXfers += targets.length;
               }
@@ -1973,9 +2195,10 @@ class FSNamesystem implements FSConstants {
             Block block = it.next();
             DatanodeDescriptor targets[] = 
                       (DatanodeDescriptor[]) replicateTargetSets.get(i);
+            int numCurrentReplica = numCurrentReplicas.get(i).intValue();
             Collection<DatanodeDescriptor> containingNodes = blocksMap.get(block);
 
-            if (containingNodes.size() + targets.length >= 
+            if (numCurrentReplica + targets.length >= 
                     dir.getFileByBlock( block).getReplication() ) {
               neededReplications.remove(block);
               pendingReplications.add(block);
@@ -2060,7 +2283,8 @@ class FSNamesystem implements FSConstants {
                  it.hasNext();) {
               DatanodeDescriptor node = it.next();
               if ((forbiddenNodes == null || !forbiddenNodes.contains(node)) &&
-                  clientMachine.toString().equals(node.getHost())) {
+                  clientMachine.toString().equals(node.getHost()) &&
+                  !node.isDecommissionInProgress() && !node.isDecommissioned()) {
                 if ((node.getRemaining() >= blockSize * MIN_BLOCKS_FOR_WRITE) &&
                     (node.getXceiverCount() <= (2.0 * avgLoad))) {
                   targets.add(node);
@@ -2084,6 +2308,7 @@ class FSNamesystem implements FSConstants {
             DatanodeDescriptor node = heartbeats.get(idx);
             if ((forbiddenNodes == null || !forbiddenNodes.contains(node)) &&
                 !targets.contains(node) &&
+                !node.isDecommissionInProgress() && !node.isDecommissioned() &&
                 (node.getRemaining() >= blockSize * MIN_BLOCKS_FOR_WRITE) &&
                 (node.getXceiverCount() <= (2.0 * avgLoad))) {
               target = node;
@@ -2100,6 +2325,7 @@ class FSNamesystem implements FSConstants {
               DatanodeDescriptor node = heartbeats.get(idx);
               if ((forbiddenNodes == null || !forbiddenNodes.contains(node)) &&
                   !targets.contains(node) &&
+                  !node.isDecommissionInProgress() && !node.isDecommissioned() &&
                   node.getRemaining() >= blockSize) {
                 target = node;
                 break;

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

@@ -465,6 +465,27 @@ public class NameNode implements ClientProtocol, DatanodeProtocol, FSConstants {
     boolean isInSafeMode() {
       return namesystem.isInSafeMode();
     }
+
+    /**
+     * Set administrative commands to decommission datanodes.
+     */
+    public boolean decommission(DecommissionAction action, String[] nodes)
+                                throws IOException {
+      boolean ret = true;
+      switch (action) {
+        case DECOMMISSION_SET: // decommission datanode(s)
+          namesystem.startDecommission(nodes);
+          break;
+        case DECOMMISSION_CLEAR: // remove decommission state of a datanode
+          namesystem.stopDecommission(nodes);
+          break;
+        case DECOMMISSION_GET: // are all the node decommissioned?
+          ret = namesystem.checkDecommissioned(nodes);
+          break;
+        }
+        return ret;
+    }
+
     
     ////////////////////////////////////////////////////////////////
     // DatanodeProtocol
@@ -513,6 +534,14 @@ public class NameNode implements ClientProtocol, DatanodeProtocol, FSConstants {
         if (blocks != null) {
             return new BlockCommand(blocks);
         }
+        //
+        // See if the decommissioned node has finished moving all
+        // its datablocks to another replica. This is a loose
+        // heuristic to determine when a decommission is really over.
+        // We can probbaly do it in a seperate thread rather than making
+        // the heartbeat thread do this.
+        //
+        namesystem.checkDecommissionState(nodeReg);
         return null;
     }
 

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

@@ -0,0 +1,227 @@
+/**
+ * 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 junit.framework.TestCase;
+import java.io.*;
+import java.util.Random;
+import java.net.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSInputStream;
+import org.apache.hadoop.fs.FSOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * This class tests the decommissioning of nodes.
+ * @author Dhruba Borthakur
+ */
+public class TestDecommission extends TestCase {
+  static final long seed = 0xDEADBEEFL;
+  static final int blockSize = 8192;
+  static final int fileSize = 16384;
+  static final int numDatanodes = 4;
+
+  Random myrand = new Random();
+
+  private void writeFile(FileSystem fileSys, Path name, int repl)
+  throws IOException {
+    // create and write a file that contains three blocks of data
+    FSOutputStream stm = fileSys.createRaw(name, true, (short)repl,
+        (long)blockSize);
+    byte[] buffer = new byte[fileSize];
+    Random rand = new Random(seed);
+    rand.nextBytes(buffer);
+    stm.write(buffer);
+    stm.close();
+  }
+  
+  
+  private void checkFile(FileSystem fileSys, Path name, int repl)
+  throws IOException {
+    String[][] locations = fileSys.getFileCacheHints(name, 0, fileSize);
+    for (int idx = 0; idx < locations.length; idx++) {
+      assertEquals("Number of replicas for block" + idx,
+          Math.min(numDatanodes, repl), locations[idx].length);  
+    }
+  }
+
+  /**
+   * For blocks that reside on the nodes that are down, verify that their
+   * replication factor is 1 more than the specified one.
+   */
+  private void checkFile(FileSystem fileSys, Path name, int repl,
+                         String[] downnodes) throws IOException {
+    FSInputStream is = fileSys.openRaw(name);
+    DFSClient.DFSInputStream dis = (DFSClient.DFSInputStream) is;
+    DatanodeInfo[][] dinfo = dis.getDataNodes();
+
+    for (int blk = 0; blk < dinfo.length; blk++) { // for each block
+      int hasdown = 0;
+      DatanodeInfo[] nodes = dinfo[blk];
+      for (int j = 0; j < nodes.length; j++) {     // for each replica
+        for (int k = 0; downnodes != null && k < downnodes.length; k++) {
+          if (nodes[j].getName().equals(downnodes[k])) {
+            hasdown++;
+            System.out.println("Block " + blk + " replica " +
+                               nodes[j].getName() + " is decommissioned.");
+          }
+        }
+      }
+      System.out.println("Block " + blk + " has " + hasdown +
+                           " decommissioned replica.");
+      assertEquals("Number of replicas for block" + blk,
+            Math.min(numDatanodes, repl+hasdown), nodes.length);  
+    }
+  }
+  
+  private void cleanupFile(FileSystem fileSys, Path name) throws IOException {
+    assertTrue(fileSys.exists(name));
+    fileSys.delete(name);
+    assertTrue(!fileSys.exists(name));
+  }
+
+  private void printDatanodeReport(DatanodeInfo[] info) {
+    System.out.println("-------------------------------------------------");
+    for (int i = 0; i < info.length; i++) {
+      System.out.println(info[i].getDatanodeReport());
+      System.out.println();
+    }
+  }
+
+  /*
+   * decommission one random node.
+   */
+  private String[] decommissionNode(DFSClient client, FileSystem filesys)
+                                    throws IOException {
+    DistributedFileSystem dfs = (DistributedFileSystem) filesys;
+    DatanodeInfo[] info = client.datanodeReport();
+
+    //
+    // pick one datanode randomly.
+    //
+    int index = myrand.nextInt(info.length);
+    String nodename = info[index].getName();
+    System.out.println("Decommissioning node: " + nodename);
+    String[] nodes = new String[1];
+    nodes[0] = nodename;
+    dfs.decommission(FSConstants.DecommissionAction.DECOMMISSION_SET, nodes);
+    return nodes;
+  }
+
+  /*
+   * put node back in action
+   */
+  private void commissionNode(DFSClient client, FileSystem filesys,
+                              String[] nodes) throws IOException {
+    DistributedFileSystem dfs = (DistributedFileSystem) filesys;
+    DatanodeInfo[] info = client.datanodeReport();
+
+    for (int i = 0; i < nodes.length; i++) {
+      System.out.println("Putting node back in action: " + nodes[i]);
+    }
+    dfs.decommission(FSConstants.DecommissionAction.DECOMMISSION_CLEAR, nodes);
+  }
+
+  /* 
+   * Check that node(s) were decommissioned
+   */
+  private void checkNodeDecommission(DFSClient client, FileSystem filesys,
+                                     String[] nodes) throws IOException {
+    DistributedFileSystem dfs = (DistributedFileSystem) filesys;
+    boolean ret = dfs.decommission(
+                    FSConstants.DecommissionAction.DECOMMISSION_GET, nodes);
+    assertEquals("State of Decommissioned Datanode(s) ", ret, true);
+  }
+
+  /* 
+   * Wait till node is fully decommissioned.
+   */
+  private void waitNodeDecommission(DFSClient client, FileSystem filesys,
+                                     String[] nodes) throws IOException {
+    DistributedFileSystem dfs = (DistributedFileSystem) filesys;
+    boolean done = dfs.decommission(
+                     FSConstants.DecommissionAction.DECOMMISSION_GET, nodes);
+    while (!done) {
+      System.out.println("Waiting for nodes " + nodes[0] +
+                         " to be fully decommissioned...");
+      try {
+        Thread.sleep(5000L);
+      } catch (InterruptedException e) {
+        // nothing
+      }
+      done = dfs.decommission(FSConstants.DecommissionAction.DECOMMISSION_GET,
+                              nodes);
+    }
+    //
+    // sleep an additional 10 seconds for the blockreports from the datanodes
+    // to arrive. 
+    //
+    try {
+      Thread.sleep(10 * 1000L);
+    } catch (Exception e) {
+    }
+  }
+  
+  /**
+   * Tests Decommission in DFS.
+   */
+  public void testDecommission() throws IOException {
+    Configuration conf = new Configuration();
+    MiniDFSCluster cluster = new MiniDFSCluster(65312, conf, numDatanodes, false);
+    // Now wait for 15 seconds to give datanodes chance to register
+    // themselves and to report heartbeat
+    try {
+      Thread.sleep(15000L);
+    } catch (InterruptedException e) {
+      // nothing
+    }
+    InetSocketAddress addr = new InetSocketAddress("localhost", 
+                                             cluster.getNameNodePort());
+    DFSClient client = new DFSClient(addr, conf);
+    DatanodeInfo[] info = client.datanodeReport();
+    assertEquals("Number of Datanodes ", numDatanodes, info.length);
+    FileSystem fileSys = cluster.getFileSystem();
+    DistributedFileSystem dfs = (DistributedFileSystem) fileSys;
+
+    try {
+      for (int iteration = 0; iteration < 2; iteration++) {
+        //
+        // Decommission one node. Verify that node is decommissioned.
+        // Verify that replication factor of file has increased from 3
+        // to 4. This means one replica is on decommissioned node.
+        // 
+        Path file1 = new Path("smallblocktest.dat");
+        writeFile(fileSys, file1, 3);
+        checkFile(fileSys, file1, 3);
+        String downnodes[] = decommissionNode(client, fileSys);
+        waitNodeDecommission(client, fileSys, downnodes);
+        checkFile(fileSys, file1, 3, downnodes);
+        commissionNode(client, fileSys, downnodes);
+        cleanupFile(fileSys, file1);
+      }
+    } catch (IOException e) {
+      info = client.datanodeReport();
+      printDatanodeReport(info);
+      throw e;
+    } finally {
+      fileSys.close();
+      cluster.shutdown();
+    }
+  }
+}

src/webapps/dfs/dfshealth.jsp

@@ -82,9 +82,15 @@
       percentUsed = FsShell.limitDecimal(((1.0 * u)/c)*100, 2);
     else
       percentUsed = "100";
+
+    String adminState = (d.isDecommissioned() ? "Decommissioned" :
+                         (d.isDecommissionInProgress() ? "Decommission In Progress":
+                          "In Service"));
     
     out.print("<td class=\"lastcontact\"> " +
               ((currentTime - d.getLastUpdate())/1000) +
+	      "<td class=\"adminstate\">" +
+              adminState +
 	      "<td class=\"size\">" +
               FsShell.limitDecimal(c*1.0/diskBytes, 2) +
 	      "<td class=\"pcused\">" + percentUsed +
@@ -167,6 +173,7 @@
 	    out.print( "<tr class=\"headerRow\"> <th " +
                        NodeHeaderStr("name") + "> Node <th " +
                        NodeHeaderStr("lastcontact") + "> Last Contact <th " +
+                       NodeHeaderStr("adminstate") + "> Admin State <th " +
                        NodeHeaderStr("size") + "> Size (" + diskByteStr +
                        ") <th " + NodeHeaderStr("pcused") +
                        "> Used (%) <th " + NodeHeaderStr("blocks") +

src/webapps/static/hadoop.css

@@ -41,7 +41,7 @@ div#dfsnodetable th.header:hover, th.headerASC:hover, th.headerDSC:hover,
 	cursor : pointer;
 }
 
-div#dfsnodetable td.blocks, td.size, td.pcused, td.lastcontact {
+div#dfsnodetable td.blocks, td.size, td.pcused, td.adminstate, td.lastcontact {
 	text-align : right;
 }