HDFS-1645. DatanodeCommond.Finalize needs to include BlockPoolId. Contributed by Suresh Srinvias.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/branches/HDFS-1052@1074679 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -31,6 +31,9 @@ Trunk (unreleased changes)
     HDFS-1653. Federation: Block received message from datanode sends invalid 
     DatanodeRegistration. (Tanping via suresh)
 
+    HDFS-1645. DatanodeCommond.Finalize needs to include BlockPoolId.
+    (suresh)
+
   IMPROVEMENTS
 
     HDFS-1510. Added test-patch.properties required by test-patch.sh (nigel)

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

@@ -1014,8 +1014,8 @@ public class DataNode extends Configured
         }
         break;
       case DatanodeProtocol.DNA_FINALIZE:
-        // TODO:FEDERATION - global storage????? or per BP storage - add real BPID
-        storage.finalizeUpgrade("FAKE ID NEEDS TO BE REPLACED");
+        storage.finalizeUpgrade(((DatanodeCommand.Finalize) cmd)
+            .getBlockPoolId());
         break;
       case UpgradeCommand.UC_ACTION_START_UPGRADE:
         // start distributed upgrade here

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

@@ -251,7 +251,7 @@ public class NameNode implements NamenodeProtocols, FSConstants {
   public static URI [] getNameNodesURIs(Configuration conf) {
     String [] nnURIs = conf.getStrings(DFSConfigKeys.DFS_FEDERATION_NAMENODES);
     if(nnURIs == null) {
-      nnURIs = new String[] { conf.get(CommonConfigurationKeys.FS_DEFAULT_NAME_KEY)};
+      nnURIs = new String[] {conf.get(CommonConfigurationKeys.FS_DEFAULT_NAME_KEY)};
     }
 
     AbstractList<URI> nns = new ArrayList<URI>(nnURIs.length);
@@ -300,6 +300,7 @@ public class NameNode implements NamenodeProtocols, FSConstants {
   /**
    * TODO:FEDERATION
    * @param filesystemURI
+   * @return address of file system
    */
   public static InetSocketAddress getAddress(URI filesystemURI) {
     String authority = filesystemURI.getAuthority();
@@ -1300,7 +1301,7 @@ public class NameNode implements NamenodeProtocols, FSConstants {
 
     namesystem.processReport(nodeReg, poolId, blist);
     if (getFSImage().isUpgradeFinalized())
-      return DatanodeCommand.FINALIZE;
+      return new DatanodeCommand.Finalize(poolId);
     return null;
   }
 
@@ -1356,8 +1357,12 @@ public class NameNode implements NamenodeProtocols, FSConstants {
    */
   public void verifyRequest(NodeRegistration nodeReg) throws IOException {
     verifyVersion(nodeReg.getVersion());
-    if (!namesystem.getRegistrationID().equals(nodeReg.getRegistrationID()))
+    if (!namesystem.getRegistrationID().equals(nodeReg.getRegistrationID())) {
+      LOG.warn("Invalid registrationID - expected: "
+          + namesystem.getRegistrationID() + " received: "
+          + nodeReg.getRegistrationID());
       throw new UnregisteredNodeException(nodeReg);
+    }
   }
     
   /**

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

@@ -19,12 +19,14 @@ package org.apache.hadoop.hdfs.server.protocol;
 
 import java.io.DataInput;
 import java.io.DataOutput;
+import java.io.IOException;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableFactory;
 import org.apache.hadoop.io.WritableFactories;
+import org.apache.hadoop.io.WritableUtils;
 import org.apache.avro.reflect.Union;
 
 /**
@@ -47,10 +49,27 @@ public abstract class DatanodeCommand extends ServerCommand {
     public void write(DataOutput out) {}
   }
 
-  static class Finalize extends DatanodeCommand {
-    private Finalize() {super(DatanodeProtocol.DNA_FINALIZE);}
-    public void readFields(DataInput in) {}
-    public void write(DataOutput out) {}
+  public static class Finalize extends DatanodeCommand {
+    String blockPoolId;
+    private Finalize() {
+      super(DatanodeProtocol.DNA_FINALIZE);
+    }
+    
+    public Finalize(String bpid) {
+      super(DatanodeProtocol.DNA_FINALIZE);
+      blockPoolId = bpid;
+    }
+    
+    public String getBlockPoolId() {
+      return blockPoolId;
+    }
+    
+    public void readFields(DataInput in) throws IOException {
+      blockPoolId = WritableUtils.readString(in);
+    }
+    public void write(DataOutput out) throws IOException {
+      WritableUtils.writeString(out, blockPoolId);
+    }
   }
 
   static {                                      // register a ctor
@@ -65,7 +84,6 @@ public abstract class DatanodeCommand extends ServerCommand {
   }
 
   public static final DatanodeCommand REGISTER = new Register();
-  public static final DatanodeCommand FINALIZE = new Finalize();
   
   public DatanodeCommand() {
     super();