AMBARI-8771. Add support for deploying HDFS NameNode HA Clusters with Blueprints. (rnettleton)

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/BaseBlueprintProcessor.java

@@ -366,6 +366,13 @@ public abstract class BaseBlueprintProcessor extends AbstractControllerResourceP
 
     Collection<String> cardinalityFailures = new HashSet<String>();
 
+    if (BlueprintConfigurationProcessor.isNameNodeHAEnabled(clusterConfig) &&
+      (component.equals("SECONDARY_NAMENODE"))) {
+      // override the cardinality for this component in an HA deployment,
+      // since the SECONDARY_NAMENODE should not be started in this scenario
+      cardinality = new Cardinality("0");
+    }
+
     int actualCount = getHostGroupsForComponent(component, hostGroups).size();
     if (! cardinality.isValidCount(actualCount)) {
       boolean validated = ! isDependencyManaged(stack, component, clusterConfig);

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/BlueprintConfigurationProcessor.java

@@ -24,6 +24,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -134,6 +135,36 @@ public class BlueprintConfigurationProcessor {
         }
       }
     }
+
+    if (isNameNodeHAEnabled()) {
+      // if the active/stanbdy namenodes are not specified, assign them automatically
+      if (! isNameNodeHAInitialActiveNodeSet(properties) && ! isNameNodeHAInitialStandbyNodeSet(properties)) {
+        Collection<HostGroup> listOfHostGroups = new LinkedList<HostGroup>();
+        for (String key : hostGroups.keySet()) {
+          listOfHostGroups.add(hostGroups.get(key));
+        }
+
+        Collection<HostGroup> hostGroupsContainingNameNode =
+          getHostGroupsForComponent("NAMENODE", listOfHostGroups);
+        // set the properties that configure which namenode is active,
+        // and which is a standby node in this HA deployment
+        Map<String, String> hadoopEnv = properties.get("hadoop-env");
+        if (hostGroupsContainingNameNode.size() == 2) {
+          List<HostGroup> listOfGroups = new LinkedList<HostGroup>(hostGroupsContainingNameNode);
+          hadoopEnv.put("dfs_ha_initial_namenode_active", listOfGroups.get(0).getHostInfo().iterator().next());
+          hadoopEnv.put("dfs_ha_initial_namenode_standby", listOfGroups.get(1).getHostInfo().iterator().next());
+        } else {
+          // handle the case where multiple hosts are mapped to an HA host group
+          if (hostGroupsContainingNameNode.size() == 1) {
+            List<String> listOfInfo = new LinkedList<String>(hostGroupsContainingNameNode.iterator().next().getHostInfo());
+            // there should only be two host names that can include a NameNode install/deployment
+            hadoopEnv.put("dfs_ha_initial_namenode_active", listOfInfo.get(0));
+            hadoopEnv.put("dfs_ha_initial_namenode_standby", listOfInfo.get(1));
+          }
+        }
+      }
+    }
+
     return properties;
   }
 
@@ -275,7 +306,45 @@ public class BlueprintConfigurationProcessor {
    *         false if NameNode HA is not enabled
    */
   boolean isNameNodeHAEnabled() {
-    return properties.containsKey("hdfs-site") && properties.get("hdfs-site").containsKey("dfs.nameservices");
+    return isNameNodeHAEnabled(properties);
+  }
+
+  /**
+   * Static convenience function to determine if NameNode HA is enabled
+   * @param configProperties configuration properties for this cluster
+   * @return true if NameNode HA is enabled
+   *         false if NameNode HA is not enabled
+   */
+  static boolean isNameNodeHAEnabled(Map<String, Map<String, String>> configProperties) {
+    return configProperties.containsKey("hdfs-site") && configProperties.get("hdfs-site").containsKey("dfs.nameservices");
+  }
+
+
+  /**
+   * Convenience method to examine the current configuration, to determine
+   * if the hostname of the initial active namenode in an HA deployment has
+   * been included.
+   *
+   * @param configProperties the configuration for this cluster
+   * @return true if the initial active namenode property has been configured
+   *         false if the initial active namenode property has not been configured
+   */
+  static boolean isNameNodeHAInitialActiveNodeSet(Map<String, Map<String, String>> configProperties) {
+    return configProperties.containsKey("hadoop-env") && configProperties.get("hadoop-env").containsKey("dfs_ha_initial_namenode_active");
+  }
+
+
+  /**
+   * Convenience method to examine the current configuration, to determine
+   * if the hostname of the initial standby namenode in an HA deployment has
+   * been included.
+   *
+   * @param configProperties the configuration for this cluster
+   * @return true if the initial standby namenode property has been configured
+   *         false if the initial standby namenode property has not been configured
+   */
+  static boolean isNameNodeHAInitialStandbyNodeSet(Map<String, Map<String, String>> configProperties) {
+    return configProperties.containsKey("hadoop-env") && configProperties.get("hadoop-env").containsKey("dfs_ha_initial_namenode_standby");
   }
 
 
@@ -456,7 +525,7 @@ public class BlueprintConfigurationProcessor {
   private static Collection<HostGroup> getHostGroupsForComponent(String component,
                                                                  Collection<? extends HostGroup> hostGroups) {
 
-    Collection<HostGroup> resultGroups = new HashSet<HostGroup>();
+    Collection<HostGroup> resultGroups = new LinkedHashSet<HostGroup>();
     for (HostGroup group : hostGroups ) {
       if (group.getComponents().contains(component)) {
         resultGroups.add(group);
@@ -476,7 +545,7 @@ public class BlueprintConfigurationProcessor {
   private static Collection<String> getHostStrings(Map<String, ? extends HostGroup> hostGroups,
                                                    String val) {
 
-    Collection<String> hosts = new HashSet<String>();
+    Collection<String> hosts = new LinkedHashSet<String>();
     Matcher m = HOSTGROUP_PORT_REGEX.matcher(val);
     while (m.find()) {
       String groupName = m.group(1);
@@ -593,6 +662,15 @@ public class BlueprintConfigurationProcessor {
           if (matchingGroups.isEmpty() && cardinality.isValidCount(0)) {
             return origValue;
           } else {
+            if (isNameNodeHAEnabled(properties) && isComponentNameNode() && (matchingGroups.size() == 2)) {
+              // if this is the defaultFS property, it should reflect the nameservice name,
+              // rather than a hostname (used in non-HA scenarios)
+              if (properties.get("core-site").get("fs.defaultFS").equals(origValue)) {
+                return origValue;
+              }
+
+            }
+
             throw new IllegalArgumentException("Unable to update configuration property with topology information. " +
               "Component '" + component + "' is not mapped to any host group or is mapped to multiple groups.");
           }
@@ -600,6 +678,17 @@ public class BlueprintConfigurationProcessor {
       }
     }
 
+    /**
+     * Utility method to determine if the component associated with this updater
+     * instance is an HDFS NameNode
+     *
+     * @return true if the component associated is a NameNode
+     *         false if the component is not a NameNode
+     */
+    private boolean isComponentNameNode() {
+      return component.equals("NAMENODE");
+    }
+
     /**
      * Provides access to the name of the component associated
      *   with this updater instance.
@@ -684,15 +773,19 @@ public class BlueprintConfigurationProcessor {
    * value with the host names which runs the associated component in the new cluster.
    */
   private static class MultipleHostTopologyUpdater implements PropertyUpdater {
+
+
+    private static final Character DEFAULT_SEPARATOR = ',';
+
     /**
      * Component name
      */
-    private String component;
+    private final String component;
 
     /**
      * Separator for multiple property values
      */
-    private Character separator = ',';
+    private final Character separator;
 
     /**
      * Constructor.
@@ -700,7 +793,19 @@ public class BlueprintConfigurationProcessor {
      * @param component  component name associated with the property
      */
     public MultipleHostTopologyUpdater(String component) {
+      this(component, DEFAULT_SEPARATOR);
+    }
+
+    /**
+     * Constructor
+     *
+     * @param component component name associated with this property
+     * @param separator the separator character to use when multiple hosts
+     *                  are specified in a property or URL
+     */
+    public MultipleHostTopologyUpdater(String component, Character separator) {
       this.component = component;
+      this.separator = separator;
     }
 
     /**
@@ -741,6 +846,30 @@ public class BlueprintConfigurationProcessor {
       }
 
       StringBuilder sb = new StringBuilder();
+      String suffix = null;
+      // parse out prefix if one exists
+      Matcher matcher = HOSTGROUP_PORT_REGEX.matcher(origValue);
+      if (matcher.find()) {
+        int indexOfStart = matcher.start();
+        // handle the case of a YAML config property
+        if ((indexOfStart > 0) && (!origValue.substring(0, indexOfStart).equals("['"))) {
+          // append prefix before adding host names
+          sb.append(origValue.substring(0, indexOfStart));
+        }
+
+        // parse out suffix if one exists
+        int indexOfEnd = -1;
+        while (matcher.find()) {
+          indexOfEnd = matcher.end();
+        }
+
+        if (indexOfEnd < (origValue.length() - 1)) {
+          suffix = origValue.substring(indexOfEnd);
+        }
+
+      }
+
+      // add hosts to property, using the specified separator
       boolean firstHost = true;
       for (String host : hostStrings) {
         if (!firstHost) {
@@ -751,6 +880,11 @@ public class BlueprintConfigurationProcessor {
         sb.append(host);
       }
 
+      if ((suffix != null) && (!suffix.equals("']"))) {
+        sb.append(suffix);
+      }
+
+
       return sb.toString();
     }
   }
@@ -961,7 +1095,8 @@ public class BlueprintConfigurationProcessor {
     hdfsSiteMap.put("dfs.namenode.https-address", new SingleHostTopologyUpdater("NAMENODE"));
     coreSiteMap.put("fs.defaultFS", new SingleHostTopologyUpdater("NAMENODE"));
     hbaseSiteMap.put("hbase.rootdir", new SingleHostTopologyUpdater("NAMENODE"));
-    multiHdfsSiteMap.put("dfs.namenode.shared.edits.dir", new MultipleHostTopologyUpdater("JOURNALNODE"));
+    // HDFS shared.edits JournalNode Quorum URL uses semi-colons as separators
+    multiHdfsSiteMap.put("dfs.namenode.shared.edits.dir", new MultipleHostTopologyUpdater("JOURNALNODE", ';'));
 
     // SECONDARY_NAMENODE
     hdfsSiteMap.put("dfs.secondary.http.address", new SingleHostTopologyUpdater("SECONDARY_NAMENODE"));

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/scripts/hdfs_namenode.py

@@ -47,6 +47,15 @@ def namenode(action=None, do_format=True, rolling_restart=False, env=None):
               group=params.user_group
     )
 
+    if params.dfs_ha_enabled:
+      # if the current host is the standby NameNode in an HA deployment
+      if params.hostname == params.dfs_ha_namenode_standby:
+        # run the bootstrap command, to start the NameNode in standby mode
+        # this requires that the active NameNode is already up and running,
+        # so this execute should be re-tried upon failure, up to a timeout
+        Execute("hdfs namenode -bootstrapStandby",
+          user = params.hdfs_user, tries=50)
+
     options = "-rollingUpgrade started" if rolling_restart else ""
 
     if rolling_restart:    
@@ -169,6 +178,23 @@ def format_namenode(force=None):
       Directory(mark_dir,
         recursive = True
       )
+  else:
+    if params.dfs_ha_namenode_active is not None:
+      if params.hostname == params.dfs_ha_namenode_active:
+        # check and run the format command in the HA deployment scenario
+        # only format the "active" namenode in an HA deployment
+        File(format("{tmp_dir}/checkForFormat.sh"),
+             content=StaticFile("checkForFormat.sh"),
+             mode=0755)
+        Execute(format(
+          "{tmp_dir}/checkForFormat.sh {hdfs_user} {hadoop_conf_dir} "
+          "{hadoop_bin_dir} {old_mark_dir} {mark_dir} {dfs_name_dir}"),
+                not_if=format("test -d {old_mark_dir} || test -d {mark_dir}"),
+                path="/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin"
+        )
+        Directory(mark_dir,
+                  recursive=True
+        )
 
 
 def decommission():

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/scripts/params.py

@@ -185,6 +185,11 @@ dfs_ha_nameservices = default("/configurations/hdfs-site/dfs.nameservices", None
 dfs_ha_namenode_ids = default(format("/configurations/hdfs-site/dfs.ha.namenodes.{dfs_ha_nameservices}"), None)
 dfs_ha_automatic_failover_enabled = default("/configurations/hdfs-site/dfs.ha.automatic-failover.enabled", False)
 
+# hostname of the active HDFS HA Namenode (only used when HA is enabled)
+dfs_ha_namenode_active = default("/configurations/hadoop-env/dfs_ha_initial_namenode_active", None)
+# hostname of the standby HDFS HA Namenode (only used when HA is enabled)
+dfs_ha_namenode_standby = default("/configurations/hadoop-env/dfs_ha_initial_namenode_standby", None)
+
 namenode_id = None
 namenode_rpc = None
 

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/scripts/zkfc_slave.py

@@ -41,6 +41,14 @@ class ZkfcSlave(Script):
               owner=params.hdfs_user,
               group=params.user_group
     )
+
+    # format the znode for this HA setup
+    # only run this format command if the active namenode hostname is set
+    # The Ambari UI HA Wizard prompts the user to run this command
+    # manually, so this guarantees it is only run in the Blueprints case
+    if params.dfs_ha_enabled and params.dfs_ha_namenode_active is not None:
+        Execute("hdfs zkfc -formatZK -force -nonInteractive", user=params.hdfs_user)
+
     utils.service(
       action="start", name="zkfc", user=params.hdfs_user, create_pid_dir=True,
       create_log_dir=True

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/BaseBlueprintProcessorTest.java

@@ -3,6 +3,7 @@ package org.apache.ambari.server.controller.internal;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.isA;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -10,9 +11,28 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.ambari.server.api.services.AmbariMetaInfo;
 import org.apache.ambari.server.controller.AmbariManagementController;
+import org.apache.ambari.server.controller.StackConfigurationResponse;
+import org.apache.ambari.server.controller.StackServiceComponentResponse;
 import org.apache.ambari.server.controller.StackServiceResponse;
+import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
+import org.apache.ambari.server.controller.spi.NoSuchResourceException;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.RequestStatus;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
+import org.apache.ambari.server.controller.spi.SystemException;
+import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
+import org.apache.ambari.server.orm.entities.BlueprintConfigEntity;
+import org.apache.ambari.server.orm.entities.BlueprintEntity;
+import org.apache.ambari.server.orm.entities.HostGroupComponentEntity;
+import org.apache.ambari.server.orm.entities.HostGroupConfigEntity;
+import org.apache.ambari.server.orm.entities.HostGroupEntity;
+import org.apache.ambari.server.state.ComponentInfo;
 import org.apache.ambari.server.state.DependencyInfo;
+import org.apache.ambari.server.state.ServiceInfo;
 import org.easymock.EasyMockSupport;
 import org.junit.Before;
 import org.junit.Test;
@@ -609,6 +629,226 @@ public class BaseBlueprintProcessorTest {
     mockSupport.verifyAll();
   }
 
+
+  @Test
+  public void testValidationOverrideForSecondaryNameNodeWithHA() throws Exception {
+    EasyMockSupport mockSupport = new EasyMockSupport();
+
+    AmbariManagementController mockController =
+      mockSupport.createMock(AmbariManagementController.class);
+
+    AmbariMetaInfo mockMetaInfo =
+      mockSupport.createMock(AmbariMetaInfo.class);
+
+    BaseBlueprintProcessor.stackInfo = mockMetaInfo;
+
+    ServiceInfo serviceInfo = new ServiceInfo();
+    serviceInfo.setName("HDFS");
+
+    StackServiceResponse stackServiceResponse =
+      new StackServiceResponse(serviceInfo);
+
+    ComponentInfo componentInfo = new ComponentInfo();
+    componentInfo.setName("SECONDARY_NAMENODE");
+    // simulate the stack requirements that there
+    // always be one SECONDARY_NAMENODE per cluster
+    componentInfo.setCardinality("1");
+
+    StackServiceComponentResponse stackComponentResponse =
+      new StackServiceComponentResponse(componentInfo);
+
+    ComponentInfo componentInfoNameNode = new ComponentInfo();
+    componentInfoNameNode.setName("NAMENODE");
+    componentInfo.setCardinality("1-2");
+    StackServiceComponentResponse stackServiceComponentResponseTwo =
+      new StackServiceComponentResponse(componentInfoNameNode);
+
+    Set<StackServiceComponentResponse> responses =
+      new HashSet<StackServiceComponentResponse>();
+    responses.add(stackComponentResponse);
+    responses.add(stackServiceComponentResponseTwo);
+
+    expect(mockController.getStackServices(isA(Set.class))).andReturn(
+      Collections.singleton(stackServiceResponse));
+    expect(mockController.getStackComponents(isA(Set.class))).andReturn(
+      responses);
+    expect(mockController.getStackConfigurations(isA(Set.class))).andReturn(Collections.<StackConfigurationResponse>emptySet());
+    expect(mockController.getStackLevelConfigurations(isA(Set.class))).andReturn(Collections.<StackConfigurationResponse>emptySet());
+
+    expect(mockMetaInfo.getComponentDependencies("HDP", "2.0.6", "HDFS", "SECONDARY_NAMENODE")).andReturn(Collections.<DependencyInfo>emptyList());
+    expect(mockMetaInfo.getComponentDependencies("HDP", "2.0.6", "HDFS", "NAMENODE")).andReturn(Collections.<DependencyInfo>emptyList());
+
+
+    mockSupport.replayAll();
+
+    BaseBlueprintProcessor baseBlueprintProcessor =
+      new BaseBlueprintProcessor(Collections.<String>emptySet(), Collections.<Resource.Type, String>emptyMap(), mockController) {
+        @Override
+        protected Set<String> getPKPropertyIds() {
+          return null;
+        }
+
+        @Override
+        public RequestStatus createResources(Request request) throws SystemException, UnsupportedPropertyException, ResourceAlreadyExistsException, NoSuchParentResourceException {
+          return null;
+        }
+
+        @Override
+        public Set<Resource> getResources(Request request, Predicate predicate) throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+          return null;
+        }
+
+        @Override
+        public RequestStatus updateResources(Request request, Predicate predicate) throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+          return null;
+        }
+
+        @Override
+        public RequestStatus deleteResources(Predicate predicate) throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+          return null;
+        }
+      };
+
+    HostGroupComponentEntity hostGroupComponentEntity =
+      new HostGroupComponentEntity();
+    // don't include the SECONDARY_NAMENODE in this entity
+    hostGroupComponentEntity.setName("NAMENODE");
+
+    HostGroupEntity hostGroupEntity =
+      new HostGroupEntity();
+    hostGroupEntity.setName("host-group-one");
+    hostGroupEntity.setComponents(Collections.singleton(hostGroupComponentEntity));
+    hostGroupEntity.setConfigurations(Collections.<HostGroupConfigEntity>emptyList());
+
+    // setup config entity to simulate the case of NameNode HA being enabled
+    BlueprintConfigEntity configEntity =
+      new BlueprintConfigEntity();
+    configEntity.setConfigData("{\"dfs.nameservices\":\"mycluster\",\"key4\":\"value4\"}");
+    configEntity.setType("hdfs-site");
+
+    BlueprintEntity testEntity =
+      new BlueprintEntity();
+    testEntity.setBlueprintName("test-blueprint");
+    testEntity.setStackName("HDP");
+    testEntity.setStackVersion("2.0.6");
+    testEntity.setHostGroups(Collections.singleton(hostGroupEntity));
+    testEntity.setConfigurations(Collections.singleton(configEntity));
+
+    baseBlueprintProcessor.validateTopology(testEntity);
+
+    mockSupport.verifyAll();
+  }
+
+  @Test
+  public void testValidationOverrideForSecondaryNameNodeWithoutHA() throws Exception {
+    EasyMockSupport mockSupport = new EasyMockSupport();
+
+    AmbariManagementController mockController =
+      mockSupport.createMock(AmbariManagementController.class);
+
+    AmbariMetaInfo mockMetaInfo =
+      mockSupport.createMock(AmbariMetaInfo.class);
+
+    BaseBlueprintProcessor.stackInfo = mockMetaInfo;
+
+    ServiceInfo serviceInfo = new ServiceInfo();
+    serviceInfo.setName("HDFS");
+
+    StackServiceResponse stackServiceResponse =
+      new StackServiceResponse(serviceInfo);
+
+    ComponentInfo componentInfo = new ComponentInfo();
+    componentInfo.setName("SECONDARY_NAMENODE");
+    // simulate the stack requirements that there
+    // always be one SECONDARY_NAMENODE per cluster
+    componentInfo.setCardinality("1");
+
+    StackServiceComponentResponse stackComponentResponse =
+      new StackServiceComponentResponse(componentInfo);
+
+    ComponentInfo componentInfoNameNode = new ComponentInfo();
+    componentInfoNameNode.setName("NAMENODE");
+    componentInfo.setCardinality("1-2");
+    StackServiceComponentResponse stackServiceComponentResponseTwo =
+      new StackServiceComponentResponse(componentInfoNameNode);
+
+    Set<StackServiceComponentResponse> responses =
+      new HashSet<StackServiceComponentResponse>();
+    responses.add(stackComponentResponse);
+    responses.add(stackServiceComponentResponseTwo);
+
+    expect(mockController.getStackServices(isA(Set.class))).andReturn(
+      Collections.singleton(stackServiceResponse));
+    expect(mockController.getStackComponents(isA(Set.class))).andReturn(
+      responses);
+    expect(mockController.getStackConfigurations(isA(Set.class))).andReturn(Collections.<StackConfigurationResponse>emptySet());
+    expect(mockController.getStackLevelConfigurations(isA(Set.class))).andReturn(Collections.<StackConfigurationResponse>emptySet());
+
+    expect(mockMetaInfo.getComponentDependencies("HDP", "2.0.6", "HDFS", "SECONDARY_NAMENODE")).andReturn(Collections.<DependencyInfo>emptyList());
+    expect(mockMetaInfo.getComponentDependencies("HDP", "2.0.6", "HDFS", "NAMENODE")).andReturn(Collections.<DependencyInfo>emptyList());
+
+
+    mockSupport.replayAll();
+
+    BaseBlueprintProcessor baseBlueprintProcessor =
+      new BaseBlueprintProcessor(Collections.<String>emptySet(), Collections.<Resource.Type, String>emptyMap(), mockController) {
+        @Override
+        protected Set<String> getPKPropertyIds() {
+          return null;
+        }
+
+        @Override
+        public RequestStatus createResources(Request request) throws SystemException, UnsupportedPropertyException, ResourceAlreadyExistsException, NoSuchParentResourceException {
+          return null;
+        }
+
+        @Override
+        public Set<Resource> getResources(Request request, Predicate predicate) throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+          return null;
+        }
+
+        @Override
+        public RequestStatus updateResources(Request request, Predicate predicate) throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+          return null;
+        }
+
+        @Override
+        public RequestStatus deleteResources(Predicate predicate) throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+          return null;
+        }
+      };
+
+    HostGroupComponentEntity hostGroupComponentEntity =
+      new HostGroupComponentEntity();
+    // don't include the SECONDARY_NAMENODE in this entity
+    hostGroupComponentEntity.setName("NAMENODE");
+
+    HostGroupEntity hostGroupEntity =
+      new HostGroupEntity();
+    hostGroupEntity.setName("host-group-one");
+    hostGroupEntity.setComponents(Collections.singleton(hostGroupComponentEntity));
+    hostGroupEntity.setConfigurations(Collections.<HostGroupConfigEntity>emptyList());
+
+
+
+    BlueprintEntity testEntity =
+      new BlueprintEntity();
+    testEntity.setBlueprintName("test-blueprint");
+    testEntity.setStackName("HDP");
+    testEntity.setStackVersion("2.0.6");
+    testEntity.setHostGroups(Collections.singleton(hostGroupEntity));
+    testEntity.setConfigurations(Collections.<BlueprintConfigEntity>emptyList());
+
+    try {
+      baseBlueprintProcessor.validateTopology(testEntity);
+      fail("IllegalArgumentException should have been thrown");
+    } catch (IllegalArgumentException expectedException) {
+      // expected exception
+    }
+
+    mockSupport.verifyAll();
+  }
+
   /**
    * Convenience class for easier setup/initialization of dependencies for unit
    * testing.

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/BlueprintConfigurationProcessorTest.java

@@ -31,6 +31,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Set;
 
@@ -1225,6 +1226,7 @@ public class BlueprintConfigurationProcessorTest {
   public void testDoUpdateForClusterWithNameNodeHAEnabled() throws Exception {
     final String expectedNameService = "mynameservice";
     final String expectedHostName = "c6401.apache.ambari.org";
+    final String expectedHostNameTwo = "serverTwo";
     final String expectedPortNum = "808080";
     final String expectedNodeOne = "nn1";
     final String expectedNodeTwo = "nn2";
@@ -1233,8 +1235,15 @@ public class BlueprintConfigurationProcessorTest {
     EasyMockSupport mockSupport = new EasyMockSupport();
 
     HostGroup mockHostGroupOne = mockSupport.createMock(HostGroup.class);
+    HostGroup mockHostGroupTwo = mockSupport.createMock(HostGroup.class);
 
-    expect(mockHostGroupOne.getHostInfo()).andReturn(Arrays.asList(expectedHostName, "serverTwo")).atLeastOnce();
+    Stack mockStack = mockSupport.createMock(Stack.class);
+
+    expect(mockHostGroupOne.getHostInfo()).andReturn(Arrays.asList(expectedHostName)).atLeastOnce();
+    expect(mockHostGroupTwo.getHostInfo()).andReturn(Arrays.asList(expectedHostNameTwo)).atLeastOnce();
+    expect(mockHostGroupOne.getComponents()).andReturn(Collections.singleton("NAMENODE")).atLeastOnce();
+    expect(mockHostGroupTwo.getComponents()).andReturn(Collections.singleton("NAMENODE")).atLeastOnce();
+    expect(mockStack.getCardinality("NAMENODE")).andReturn(new Cardinality("1-2")).atLeastOnce();
 
     mockSupport.replayAll();
 
@@ -1243,8 +1252,15 @@ public class BlueprintConfigurationProcessorTest {
 
     Map<String, String> hdfsSiteProperties =
       new HashMap<String, String>();
+    Map<String, String> hadoopEnvProperties =
+      new HashMap<String, String>();
+    Map<String, String> coreSiteProperties =
+      new HashMap<String, String>();
+
 
     configProperties.put("hdfs-site", hdfsSiteProperties);
+    configProperties.put("hadoop-env", hadoopEnvProperties);
+    configProperties.put("core-site", coreSiteProperties);
 
     // setup hdfs HA config for test
     hdfsSiteProperties.put("dfs.nameservices", expectedNameService);
@@ -1259,6 +1275,94 @@ public class BlueprintConfigurationProcessorTest {
     hdfsSiteProperties.put("dfs.namenode.rpc-address." + expectedNameService + "." + expectedNodeOne, createExportedAddress(expectedPortNum, expectedHostGroupName));
     hdfsSiteProperties.put("dfs.namenode.rpc-address." + expectedNameService + "." + expectedNodeTwo, createExportedAddress(expectedPortNum, expectedHostGroupName));
 
+    // configure the defaultFS to use the nameservice URL
+    coreSiteProperties.put("fs.defaultFS", "hdfs://" + expectedNameService);
+
+    BlueprintConfigurationProcessor configProcessor =
+      new BlueprintConfigurationProcessor(configProperties);
+
+    Map<String, HostGroup> mapOfHostGroups = new LinkedHashMap<String, HostGroup>();
+    mapOfHostGroups.put(expectedHostGroupName, mockHostGroupOne);
+    mapOfHostGroups.put("host-group-2", mockHostGroupTwo);
+
+    configProcessor.doUpdateForClusterCreate(mapOfHostGroups, mockStack);
+
+    // verify that the expected hostname was substituted for the host group name in the config
+    assertEquals("HTTPS address HA property not properly exported",
+      expectedHostName + ":" + expectedPortNum, hdfsSiteProperties.get("dfs.namenode.https-address." + expectedNameService + "." + expectedNodeOne));
+    assertEquals("HTTPS address HA property not properly exported",
+      expectedHostName + ":" + expectedPortNum, hdfsSiteProperties.get("dfs.namenode.https-address." + expectedNameService + "." + expectedNodeTwo));
+
+    assertEquals("HTTPS address HA property not properly exported",
+      expectedHostName + ":" + expectedPortNum, hdfsSiteProperties.get("dfs.namenode.http-address." + expectedNameService + "." + expectedNodeOne));
+    assertEquals("HTTPS address HA property not properly exported",
+      expectedHostName + ":" + expectedPortNum, hdfsSiteProperties.get("dfs.namenode.http-address." + expectedNameService + "." + expectedNodeTwo));
+
+    assertEquals("HTTPS address HA property not properly exported",
+      expectedHostName + ":" + expectedPortNum, hdfsSiteProperties.get("dfs.namenode.rpc-address." + expectedNameService + "." + expectedNodeOne));
+    assertEquals("HTTPS address HA property not properly exported",
+      expectedHostName + ":" + expectedPortNum, hdfsSiteProperties.get("dfs.namenode.rpc-address." + expectedNameService + "." + expectedNodeTwo));
+
+    // verify that the Blueprint config processor has set the internal required properties
+    // that determine the active and standby node hostnames for this HA setup
+    assertEquals("Active Namenode hostname was not set correctly",
+      expectedHostName, hadoopEnvProperties.get("dfs_ha_initial_namenode_active"));
+
+    assertEquals("Standby Namenode hostname was not set correctly",
+      expectedHostNameTwo, hadoopEnvProperties.get("dfs_ha_initial_namenode_standby"));
+
+    assertEquals("fs.defaultFS should not be modified by cluster update when NameNode HA is enabled.",
+                 "hdfs://" + expectedNameService, coreSiteProperties.get("fs.defaultFS"));
+
+    mockSupport.verifyAll();
+  }
+
+  @Test
+  public void testDoUpdateForClusterWithNameNodeHAEnabledAndActiveNodeSet() throws Exception {
+    final String expectedNameService = "mynameservice";
+    final String expectedHostName = "serverThree";
+    final String expectedHostNameTwo = "serverFour";
+    final String expectedPortNum = "808080";
+    final String expectedNodeOne = "nn1";
+    final String expectedNodeTwo = "nn2";
+    final String expectedHostGroupName = "host_group_1";
+
+    EasyMockSupport mockSupport = new EasyMockSupport();
+
+    HostGroup mockHostGroupOne = mockSupport.createMock(HostGroup.class);
+
+    expect(mockHostGroupOne.getHostInfo()).andReturn(Arrays.asList(expectedHostName, expectedHostNameTwo)).atLeastOnce();
+
+    mockSupport.replayAll();
+
+    Map<String, Map<String, String>> configProperties =
+      new HashMap<String, Map<String, String>>();
+
+    Map<String, String> hdfsSiteProperties =
+      new HashMap<String, String>();
+
+    Map<String, String> hadoopEnvProperties =
+      new HashMap<String, String>();
+
+    configProperties.put("hdfs-site", hdfsSiteProperties);
+    configProperties.put("hadoop-env", hadoopEnvProperties);
+
+    // setup hdfs HA config for test
+    hdfsSiteProperties.put("dfs.nameservices", expectedNameService);
+    hdfsSiteProperties.put("dfs.ha.namenodes.mynameservice", expectedNodeOne + ", " + expectedNodeTwo);
+
+    // setup properties that include exported host group information
+    hdfsSiteProperties.put("dfs.namenode.https-address." + expectedNameService + "." + expectedNodeOne, createExportedAddress(expectedPortNum, expectedHostGroupName));
+    hdfsSiteProperties.put("dfs.namenode.https-address." + expectedNameService + "." + expectedNodeTwo, createExportedAddress(expectedPortNum, expectedHostGroupName));
+    hdfsSiteProperties.put("dfs.namenode.http-address." + expectedNameService + "." + expectedNodeOne, createExportedAddress(expectedPortNum, expectedHostGroupName));
+    hdfsSiteProperties.put("dfs.namenode.http-address." + expectedNameService + "." + expectedNodeTwo, createExportedAddress(expectedPortNum, expectedHostGroupName));
+    hdfsSiteProperties.put("dfs.namenode.rpc-address." + expectedNameService + "." + expectedNodeOne, createExportedAddress(expectedPortNum, expectedHostGroupName));
+    hdfsSiteProperties.put("dfs.namenode.rpc-address." + expectedNameService + "." + expectedNodeTwo, createExportedAddress(expectedPortNum, expectedHostGroupName));
+
+    // set hadoop-env properties to explicitly configure the initial
+    // active and stanbdy namenodes
+    hadoopEnvProperties.put("dfs_ha_initial_namenode_active", expectedHostName);
+    hadoopEnvProperties.put("dfs_ha_initial_namenode_standby", expectedHostNameTwo);
 
     BlueprintConfigurationProcessor configProcessor =
       new BlueprintConfigurationProcessor(configProperties);
@@ -1284,6 +1388,15 @@ public class BlueprintConfigurationProcessorTest {
     assertEquals("HTTPS address HA property not properly exported",
       expectedHostName + ":" + expectedPortNum, hdfsSiteProperties.get("dfs.namenode.rpc-address." + expectedNameService + "." + expectedNodeTwo));
 
+    // verify that the Blueprint config processor has not overridden
+    // the user's configuration to determine the active and
+    // standby nodes in this NameNode HA cluster
+    assertEquals("Active Namenode hostname was not set correctly",
+      expectedHostName, hadoopEnvProperties.get("dfs_ha_initial_namenode_active"));
+
+    assertEquals("Standby Namenode hostname was not set correctly",
+      expectedHostNameTwo, hadoopEnvProperties.get("dfs_ha_initial_namenode_standby"));
+
     mockSupport.verifyAll();
   }
 
@@ -1840,6 +1953,56 @@ public class BlueprintConfigurationProcessorTest {
 
   }
 
+  @Test
+  public void testHDFSConfigClusterUpdateQuorumJournalURL() throws Exception {
+    final String expectedHostNameOne = "c6401.apache.ambari.org";
+    final String expectedHostNameTwo = "c6402.apache.ambari.org";
+    final String expectedPortNum = "808080";
+    final String expectedHostGroupName = "host_group_1";
+    final String expectedHostGroupNameTwo = "host_group_2";
+
+    EasyMockSupport mockSupport = new EasyMockSupport();
+
+    HostGroup mockHostGroupOne = mockSupport.createMock(HostGroup.class);
+    HostGroup mockHostGroupTwo = mockSupport.createMock(HostGroup.class);
+
+    expect(mockHostGroupOne.getHostInfo()).andReturn(Arrays.asList(expectedHostNameOne)).atLeastOnce();
+    expect(mockHostGroupTwo.getHostInfo()).andReturn(Arrays.asList(expectedHostNameTwo)).atLeastOnce();
+
+    mockSupport.replayAll();
+
+    Map<String, Map<String, String>> configProperties =
+      new HashMap<String, Map<String, String>>();
+
+    Map<String, String> hdfsSiteProperties =
+      new HashMap<String, String>();
+
+    configProperties.put("hdfs-site", hdfsSiteProperties);
+
+    // setup properties that include host information
+    // setup shared edit property, that includes a qjournal URL scheme
+    hdfsSiteProperties.put("dfs.namenode.shared.edits.dir", "qjournal://" + createExportedAddress(expectedPortNum, expectedHostGroupName) + ";" + createExportedAddress(expectedPortNum, expectedHostGroupNameTwo) + "/mycluster");
+
+    BlueprintConfigurationProcessor configProcessor =
+      new BlueprintConfigurationProcessor(configProperties);
+
+    Map<String, HostGroup> mapOfHostGroups =
+      new HashMap<String, HostGroup>();
+    mapOfHostGroups.put(expectedHostGroupName, mockHostGroupOne);
+    mapOfHostGroups.put(expectedHostGroupNameTwo, mockHostGroupTwo);
+
+    // call top-level export method
+    configProcessor.doUpdateForClusterCreate(mapOfHostGroups, null);
+
+    // expect that all servers are included in the updated config, and that the qjournal URL format is preserved
+    assertEquals("HDFS HA shared edits directory property not properly updated for cluster create.",
+      "qjournal://" + createHostAddress(expectedHostNameOne, expectedPortNum) + ";" + createHostAddress(expectedHostNameTwo, expectedPortNum) + "/mycluster",
+      hdfsSiteProperties.get("dfs.namenode.shared.edits.dir"));
+
+    mockSupport.verifyAll();
+
+  }
+
   @Test
   public void testHiveConfigExported() throws Exception {
     final String expectedHostName = "c6401.apache.ambari.org";

ambari-server/src/test/python/stacks/2.0.6/HDFS/test_namenode.py

@@ -435,6 +435,184 @@ class TestNamenode(RMFTestCase):
     )
     self.assertNoMoreResources()
 
+  # tests namenode start command when NameNode HA is enabled, and
+  # the HA cluster is started initially, rather than using the UI Wizard
+  def test_start_ha_bootstrap_active_from_blueprint(self):
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/namenode.py",
+                       classname = "NameNode",
+                       command = "start",
+                       config_file="ha_bootstrap_active_node.json",
+                       hdp_stack_version = self.STACK_VERSION,
+                       target = RMFTestCase.TARGET_COMMON_SERVICES
+    )
+    self.assert_configure_default()
+
+    # verify that active namenode was formatted
+    self.assertResourceCalled('File', '/tmp/checkForFormat.sh',
+                              content = StaticFile('checkForFormat.sh'),
+                              mode = 0755,
+                              )
+    self.assertResourceCalled('Execute', '/tmp/checkForFormat.sh hdfs /etc/hadoop/conf /usr/bin /var/run/hadoop/hdfs/namenode/formatted/ /var/lib/hdfs/namenode/formatted/ /hadoop/hdfs/namenode',
+                              path = ['/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin'],
+                              not_if = 'test -d /var/run/hadoop/hdfs/namenode/formatted/ || test -d /var/lib/hdfs/namenode/formatted/',
+                              )
+    self.assertResourceCalled('Directory', '/var/lib/hdfs/namenode/formatted/',
+                              recursive = True,
+                              )
+
+    self.assertResourceCalled('File', '/etc/hadoop/conf/dfs.exclude',
+                              owner = 'hdfs',
+                              content = Template('exclude_hosts_list.j2'),
+                              group = 'hadoop',
+                              )
+    self.assertResourceCalled('Directory', '/var/run/hadoop',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              mode = 0755
+    )
+    self.assertResourceCalled('Directory', '/var/run/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('Directory', '/var/log/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('File', '/var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid',
+                              action = ['delete'],
+                              not_if='ls /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid` >/dev/null 2>&1',
+                              )
+    self.assertResourceCalled('Execute', "/usr/bin/sudo su hdfs -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]ulimit -c unlimited &&  /usr/lib/hadoop/sbin/hadoop-daemon.sh --config /etc/hadoop/conf start namenode'",
+                              environment = {'HADOOP_LIBEXEC_DIR': '/usr/lib/hadoop/libexec'},
+                              not_if = 'ls /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid` >/dev/null 2>&1',
+                              )
+    self.assertResourceCalled('Execute', "hadoop dfsadmin -safemode get | grep 'Safe mode is OFF'",
+                              path = ['/usr/bin'],
+                              tries = 40,
+                              only_if = "/usr/bin/sudo su hdfs -l -s /bin/bash -c 'export  PATH=/bin:/usr/bin ; hdfs --config /etc/hadoop/conf haadmin -getServiceState nn1 | grep active'",
+                              user = 'hdfs',
+                              try_sleep = 10,
+                              )
+    self.assertResourceCalled('HdfsDirectory', '/tmp',
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = '/usr/bin/kinit',
+                              mode = 0777,
+                              owner = 'hdfs',
+                              bin_dir = '/usr/bin',
+                              action = ['create_delayed'],
+                              )
+    self.assertResourceCalled('HdfsDirectory', '/user/ambari-qa',
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = '/usr/bin/kinit',
+                              mode = 0770,
+                              owner = 'ambari-qa',
+                              bin_dir = '/usr/bin',
+                              action = ['create_delayed'],
+                              )
+    self.assertResourceCalled('HdfsDirectory', None,
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = '/usr/bin/kinit',
+                              action = ['create'],
+                              bin_dir = '/usr/bin',
+                              only_if = "/usr/bin/sudo su hdfs -l -s /bin/bash -c 'export  PATH=/bin:/usr/bin ; hdfs --config /etc/hadoop/conf haadmin -getServiceState nn1 | grep active'",
+                              )
+    self.assertNoMoreResources()
+
+  # tests namenode start command when NameNode HA is enabled, and
+  # the HA cluster is started initially, rather than using the UI Wizard
+  # this test verifies the startup of a "standby" namenode
+  def test_start_ha_bootstrap_standby_from_blueprint(self):
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/namenode.py",
+                       classname = "NameNode",
+                       command = "start",
+                       config_file="ha_bootstrap_standby_node.json",
+                       hdp_stack_version = self.STACK_VERSION,
+                       target = RMFTestCase.TARGET_COMMON_SERVICES
+    )
+    self.assert_configure_default()
+
+    self.assertResourceCalled('File', '/etc/hadoop/conf/dfs.exclude',
+                              owner = 'hdfs',
+                              content = Template('exclude_hosts_list.j2'),
+                              group = 'hadoop',
+                              )
+    self.assertResourceCalled('Directory', '/var/run/hadoop',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              mode = 0755
+    )
+
+    # verify that the standby case is detected, and that the bootstrap
+    # command is run before the namenode launches
+    self.assertResourceCalled('Execute', 'hdfs namenode -bootstrapStandby',
+                              user = 'hdfs', tries=50)
+
+    self.assertResourceCalled('Directory', '/var/run/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('Directory', '/var/log/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('File', '/var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid',
+                              action = ['delete'],
+                              not_if='ls /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid` >/dev/null 2>&1',
+                              )
+    self.assertResourceCalled('Execute', "/usr/bin/sudo su hdfs -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]ulimit -c unlimited &&  /usr/lib/hadoop/sbin/hadoop-daemon.sh --config /etc/hadoop/conf start namenode'",
+                              environment = {'HADOOP_LIBEXEC_DIR': '/usr/lib/hadoop/libexec'},
+                              not_if = 'ls /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-namenode.pid` >/dev/null 2>&1',
+                              )
+    self.assertResourceCalled('Execute', "hadoop dfsadmin -safemode get | grep 'Safe mode is OFF'",
+                              path = ['/usr/bin'],
+                              tries = 40,
+                              only_if = "/usr/bin/sudo su hdfs -l -s /bin/bash -c 'export  PATH=/bin:/usr/bin ; hdfs --config /etc/hadoop/conf haadmin -getServiceState nn2 | grep active'",
+                              user = 'hdfs',
+                              try_sleep = 10,
+                              )
+    self.assertResourceCalled('HdfsDirectory', '/tmp',
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = '/usr/bin/kinit',
+                              mode = 0777,
+                              owner = 'hdfs',
+                              bin_dir = '/usr/bin',
+                              action = ['create_delayed'],
+                              )
+    self.assertResourceCalled('HdfsDirectory', '/user/ambari-qa',
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = '/usr/bin/kinit',
+                              mode = 0770,
+                              owner = 'ambari-qa',
+                              bin_dir = '/usr/bin',
+                              action = ['create_delayed'],
+                              )
+    self.assertResourceCalled('HdfsDirectory', None,
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = '/usr/bin/kinit',
+                              action = ['create'],
+                              bin_dir = '/usr/bin',
+                              only_if = "/usr/bin/sudo su hdfs -l -s /bin/bash -c 'export  PATH=/bin:/usr/bin ; hdfs --config /etc/hadoop/conf haadmin -getServiceState nn2 | grep active'",
+                              )
+    self.assertNoMoreResources()
+
   def test_decommission_default(self):
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/namenode.py",
                        classname = "NameNode",

ambari-server/src/test/python/stacks/2.0.6/HDFS/test_zkfc.py

@@ -205,3 +205,136 @@ class TestZkfc(RMFTestCase):
                               action = ['delete'],
                               )
     self.assertNoMoreResources()
+
+
+  def test_start_with_ha_active_namenode_bootstrap(self):
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/zkfc_slave.py",
+                       classname = "ZkfcSlave",
+                       command = "start",
+                       config_file="ha_bootstrap_active_node.json",
+                       hdp_stack_version = self.STACK_VERSION,
+                       target = RMFTestCase.TARGET_COMMON_SERVICES
+    )
+    self.assertResourceCalled('Directory', '/etc/security/limits.d',
+                              owner = 'root',
+                              group = 'root',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('File', '/etc/security/limits.d/hdfs.conf',
+                              content = Template('hdfs.conf.j2'),
+                              owner = 'root',
+                              group = 'root',
+                              mode = 0644,
+                              )
+    self.assertResourceCalled('XmlConfig', 'hdfs-site.xml',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              conf_dir = '/etc/hadoop/conf',
+                              configurations = self.getConfig()['configurations']['hdfs-site'],
+                              configuration_attributes = self.getConfig()['configuration_attributes']['hdfs-site']
+    )
+    self.assertResourceCalled('XmlConfig', 'core-site.xml',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              conf_dir = '/etc/hadoop/conf',
+                              configurations = self.getConfig()['configurations']['core-site'],
+                              configuration_attributes = self.getConfig()['configuration_attributes']['core-site'],
+                              mode = 0644
+    )
+    self.assertResourceCalled('File', '/etc/hadoop/conf/slaves',
+                              content = Template('slaves.j2'),
+                              owner = 'hdfs',
+                              )
+    self.assertResourceCalled('Directory', '/var/run/hadoop',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              mode = 0755
+    )
+
+    # verify that the znode initialization occurs prior to ZKFC startup
+    self.assertResourceCalled('Execute', 'hdfs zkfc -formatZK -force -nonInteractive',
+                              user = 'hdfs')
+
+    self.assertResourceCalled('Directory', '/var/run/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('Directory', '/var/log/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('File', '/var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid',
+                              action = ['delete'],
+                              not_if = 'ls /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid` >/dev/null 2>&1',
+                              )
+    self.assertResourceCalled('Execute', "/usr/bin/sudo su hdfs -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]ulimit -c unlimited &&  /usr/lib/hadoop/sbin/hadoop-daemon.sh --config /etc/hadoop/conf start zkfc'",
+                              environment = {'HADOOP_LIBEXEC_DIR': '/usr/lib/hadoop/libexec'},
+                              not_if = 'ls /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid` >/dev/null 2>&1',
+                              )
+    self.assertNoMoreResources()
+
+  def test_start_with_ha_standby_namenode_bootstrap(self):
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/zkfc_slave.py",
+                       classname = "ZkfcSlave",
+                       command = "start",
+                       config_file="ha_bootstrap_standby_node.json",
+                       hdp_stack_version = self.STACK_VERSION,
+                       target = RMFTestCase.TARGET_COMMON_SERVICES
+    )
+    self.assertResourceCalled('Directory', '/etc/security/limits.d',
+                              owner = 'root',
+                              group = 'root',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('File', '/etc/security/limits.d/hdfs.conf',
+                              content = Template('hdfs.conf.j2'),
+                              owner = 'root',
+                              group = 'root',
+                              mode = 0644,
+                              )
+    self.assertResourceCalled('XmlConfig', 'hdfs-site.xml',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              conf_dir = '/etc/hadoop/conf',
+                              configurations = self.getConfig()['configurations']['hdfs-site'],
+                              configuration_attributes = self.getConfig()['configuration_attributes']['hdfs-site']
+    )
+    self.assertResourceCalled('XmlConfig', 'core-site.xml',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              conf_dir = '/etc/hadoop/conf',
+                              configurations = self.getConfig()['configurations']['core-site'],
+                              configuration_attributes = self.getConfig()['configuration_attributes']['core-site'],
+                              mode = 0644
+    )
+    self.assertResourceCalled('File', '/etc/hadoop/conf/slaves',
+                              content = Template('slaves.j2'),
+                              owner = 'hdfs',
+                              )
+    self.assertResourceCalled('Directory', '/var/run/hadoop',
+                              owner = 'hdfs',
+                              group = 'hadoop',
+                              mode = 0755
+    )
+
+    # verify that the znode initialization occurs prior to ZKFC startup
+    self.assertResourceCalled('Execute', 'hdfs zkfc -formatZK -force -nonInteractive',
+                              user = 'hdfs')
+
+    self.assertResourceCalled('Directory', '/var/run/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('Directory', '/var/log/hadoop/hdfs',
+                              owner = 'hdfs',
+                              recursive = True,
+                              )
+    self.assertResourceCalled('File', '/var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid',
+                              action = ['delete'],
+                              not_if = 'ls /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid` >/dev/null 2>&1',
+                              )
+    self.assertResourceCalled('Execute', "/usr/bin/sudo su hdfs -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]ulimit -c unlimited &&  /usr/lib/hadoop/sbin/hadoop-daemon.sh --config /etc/hadoop/conf start zkfc'",
+                              environment = {'HADOOP_LIBEXEC_DIR': '/usr/lib/hadoop/libexec'},
+                              not_if = 'ls /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid >/dev/null 2>&1 && ps -p `cat /var/run/hadoop/hdfs/hadoop-hdfs-zkfc.pid` >/dev/null 2>&1',
+                              )
+    self.assertNoMoreResources()