/*
 * 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.ambari.server.controller.internal;


import static java.util.stream.Collectors.groupingBy;
import static java.util.stream.Collectors.toSet;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.annotation.Nullable;
import javax.validation.constraints.NotNull;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.PropertyDependencyInfo;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.ValueAttributesInfo;
import org.apache.ambari.server.topology.AdvisedConfiguration;
import org.apache.ambari.server.topology.Blueprint;
import org.apache.ambari.server.topology.Cardinality;
import org.apache.ambari.server.topology.ClusterTopology;
import org.apache.ambari.server.topology.ConfigRecommendationStrategy;
import org.apache.ambari.server.topology.Configuration;
import org.apache.ambari.server.topology.HostGroup;
import org.apache.ambari.server.topology.HostGroupInfo;
import org.apache.ambari.server.topology.InvalidTopologyException;
import org.apache.ambari.server.topology.validators.NameNodeHaValidator;
import org.apache.ambari.server.topology.validators.UnitValidatedProperty;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicates;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

/**
 * Updates configuration properties based on cluster topology.  This is done when exporting
 * a blueprint and when a cluster is provisioned via a blueprint.
 */
public class BlueprintConfigurationProcessor {

  private static final Logger LOG = LoggerFactory.getLogger(BlueprintConfigurationProcessor.class);

  private final static String COMMAND_RETRY_ENABLED_PROPERTY_NAME = "command_retry_enabled";

  private final static String COMMANDS_TO_RETRY_PROPERTY_NAME = "commands_to_retry";

  private final static String COMMAND_RETRY_MAX_TIME_IN_SEC_PROPERTY_NAME = "command_retry_max_time_in_sec";

  private final static String COMMAND_RETRY_ENABLED_DEFAULT = "true";

  private final static String COMMANDS_TO_RETRY_DEFAULT = "INSTALL,START";

  private final static String COMMAND_RETRY_MAX_TIME_IN_SEC_DEFAULT = "600";

  private final static String CLUSTER_ENV_CONFIG_TYPE_NAME = "cluster-env";

  private final static String HBASE_SITE_HBASE_COPROCESSOR_MASTER_CLASSES = "hbase.coprocessor.master.classes";
  private final static String HBASE_SITE_HBASE_COPROCESSOR_REGION_CLASSES = "hbase.coprocessor.region.classes";

  private final static String HAWQ_SITE_HAWQ_STANDBY_ADDRESS_HOST = "hawq_standby_address_host";
  private final static String HAWQSTANDBY = "HAWQSTANDBY";

  private final static String HDFS_HA_INITIAL_CONFIG_TYPE = CLUSTER_ENV_CONFIG_TYPE_NAME;
  private final static String HDFS_ACTIVE_NAMENODE_PROPERTY_NAME = "dfs_ha_initial_namenode_active";
  private final static String HDFS_STANDBY_NAMENODE_PROPERTY_NAME = "dfs_ha_initial_namenode_standby";
  private final static String HDFS_ACTIVE_NAMENODE_SET_PROPERTY_NAME = "dfs_ha_initial_namenode_active_set";
  private final static String HDFS_STANDBY_NAMENODE_SET_PROPERTY_NAME = "dfs_ha_initial_namenode_standby_set";

  private final static String HDFS_HA_INITIAL_CLUSTER_ID_PROPERTY_NAME = "dfs_ha_initial_cluster_id";
  private final static Set<String> HDFS_HA_INITIAL_PROPERTIES = ImmutableSet.of(
      HDFS_ACTIVE_NAMENODE_PROPERTY_NAME, HDFS_ACTIVE_NAMENODE_SET_PROPERTY_NAME,
      HDFS_STANDBY_NAMENODE_PROPERTY_NAME, HDFS_STANDBY_NAMENODE_SET_PROPERTY_NAME,
      HDFS_HA_INITIAL_CLUSTER_ID_PROPERTY_NAME);

  /**
   * These properties are only required during deployment, and should be removed afterwards.
   */
  public static final Map<String, Set<String>> TEMPORARY_PROPERTIES_FOR_CLUSTER_DEPLOYMENT = ImmutableMap.of(
    HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_HA_INITIAL_PROPERTIES
  );

  private final static String HADOOP_ENV_CONFIG_TYPE_NAME = "hadoop-env";
  private final static String RANGER_TAGSYNC_SITE_CONFIG_TYPE_NAME = "ranger-tagsync-site";
  private static final String LOCALHOST = "localhost";


  /**
   * Single host topology updaters
   */
  protected static final Map<String, Map<String, PropertyUpdater>> singleHostTopologyUpdaters =
    new HashMap<>();

  /**
   * Multi host topology updaters
   */
  private static final Map<String, Map<String, PropertyUpdater>> multiHostTopologyUpdaters =
    new HashMap<>();

  /**
   * Database host topology updaters
   */
  private static final Map<String, Map<String, PropertyUpdater>> dbHostTopologyUpdaters =
    new HashMap<>();

  /**
   * Updaters for properties which need 'm' appended
   */
  private static final Map<String, Map<String, PropertyUpdater>> mPropertyUpdaters =
    new HashMap<>();

  /**
   * Non topology related updaters
   */
  private static final Map<String, Map<String, PropertyUpdater>> nonTopologyUpdaters =
    new HashMap<>();

  /**
   * Updaters that preserve the original property value, functions
   * as a placeholder for DB-related properties that need to be
   * removed from export, but do not require an update during
   * cluster creation
   */
  private final Map<String, Map<String, PropertyUpdater>> removePropertyUpdaters =
    new HashMap<>();

  /**
   * Collection of all updaters
   */
  private static final Collection<Map<String, Map<String, PropertyUpdater>>> allUpdaters =
    new ArrayList<>();

  /**
   * Compiled regex for hostgroup token with port information.
   */
  private static final Pattern HOSTGROUP_PORT_REGEX = Pattern.compile("%HOSTGROUP::(\\S+?)%:?(\\d+)?");

  /**
   * Compiled regex for hostgroup token with port information.
   */
  private static final Pattern LOCALHOST_PORT_REGEX = Pattern.compile("localhost:?(\\d+)?");

  /**
   * Compiled regex for placeholder
   */
  private static final Pattern PLACEHOLDER = Pattern.compile("\\{\\{.*\\}\\}");

  /**
   * Special network address
   */
  private static final String BIND_ALL_IP_ADDRESS = "0.0.0.0";

  /**
   * Statically-defined set of properties that can support HA using a nameservice name
   *   in the configuration, rather than just a host name.
   *   This set also contains other HA properties that will be exported if the
   *   expected hostname information is not found.
   */
  private static final Set<String> configPropertiesWithHASupport =
    new HashSet<>(Arrays.asList("fs.defaultFS", "hbase.rootdir", "instance.volumes", "policymgr_external_url", "xasecure.audit.destination.hdfs.dir"));

  private static final Set<Pair<String, String>> PROPERTIES_FOR_HADOOP_PROXYUSER = ImmutableSet.of(
    Pair.of("oozie-env", "oozie_user"),
    Pair.of("hive-env", "hive_user"),
    Pair.of("hive-env", "webhcat_user"),
    Pair.of("hbase-env", "hbase_user"),
    Pair.of("falcon-env", "falcon_user")
  );
  private static final String HADOOP_PROXYUSER_HOSTS_FORMAT = "hadoop.proxyuser.%s.hosts";
  private static final String HADOOP_PROXYUSER_GROUPS_FORMAT = "hadoop.proxyuser.%s.groups";

  /**
   * Statically-defined list of filters to apply on property exports.
   * This will initially be used to filter out the Ranger Passwords, but
   * could be extended in the future for more generic purposes.
   */
  private PropertyFilter[] getExportPropertyFilters(Map<Long, Set<String>> authToLocalPerClusterMap)
  {
    return new PropertyFilter[] {
      new PasswordPropertyFilter(),
      new SimplePropertyNameExportFilter("tez.tez-ui.history-url.base", "tez-site"),
      new SimplePropertyNameExportFilter("admin_server_host", "kerberos-env"),
      new SimplePropertyNameExportFilter("kdc_hosts", "kerberos-env"),
      new SimplePropertyNameExportFilter("master_kdc", "kerberos-env"),
      new SimplePropertyNameExportFilter("realm", "kerberos-env"),
      new SimplePropertyNameExportFilter("kdc_type", "kerberos-env"),
      new SimplePropertyNameExportFilter("ldap-url", "kerberos-env"),
      new SimplePropertyNameExportFilter("container_dn", "kerberos-env"),
      new SimplePropertyNameExportFilter("domains", "krb5-conf"),
      new SimplePropertyNameExportFilter(HDFS_ACTIVE_NAMENODE_PROPERTY_NAME, HADOOP_ENV_CONFIG_TYPE_NAME),
      new SimplePropertyNameExportFilter(HDFS_STANDBY_NAMENODE_PROPERTY_NAME, HADOOP_ENV_CONFIG_TYPE_NAME),
      new SimplePropertyNameExportFilter(HDFS_ACTIVE_NAMENODE_SET_PROPERTY_NAME, HADOOP_ENV_CONFIG_TYPE_NAME),
      new SimplePropertyNameExportFilter(HDFS_STANDBY_NAMENODE_SET_PROPERTY_NAME, HADOOP_ENV_CONFIG_TYPE_NAME),
      new SimplePropertyNameExportFilter(HDFS_ACTIVE_NAMENODE_PROPERTY_NAME, HDFS_HA_INITIAL_CONFIG_TYPE),
      new SimplePropertyNameExportFilter(HDFS_STANDBY_NAMENODE_PROPERTY_NAME, HDFS_HA_INITIAL_CONFIG_TYPE),
      new SimplePropertyNameExportFilter(HDFS_ACTIVE_NAMENODE_SET_PROPERTY_NAME, HDFS_HA_INITIAL_CONFIG_TYPE),
      new SimplePropertyNameExportFilter(HDFS_STANDBY_NAMENODE_SET_PROPERTY_NAME, HDFS_HA_INITIAL_CONFIG_TYPE),
      new StackPropertyTypeFilter(),
      new KerberosAuthToLocalRulesFilter(authToLocalPerClusterMap)};
  }

  /**
   * Statically-defined list of filters to apply on cluster config
   * property updates.
   *
   * This will initially be used to filter out properties that do not
   * need to be set, due to a given dependency property not having
   * an expected value.
   *
   * The UI uses the Recommendations/StackAdvisor APIs to accomplish this, but
   * Blueprints will use filters in the short-term, and hopefully move to a more
   * unified approach in the next release.
   *
   * This filter approach will also be used to remove properties in a given component
   * that are not valid in a High-Availability deployment (example: HDFS NameNode HA).
   */
  private static final PropertyFilter[] clusterUpdatePropertyFilters =
    { new DependencyEqualsFilter("hbase.security.authorization", "hbase-site", "true"),
      new DependencyNotEqualsFilter("hive.server2.authentication", "hive-site", "NONE"),
      /* Temporary solution related to HBASE/Phoenix issue PHOENIX-3360, to remove hbase.rpc.controllerfactory
       * .class from hbase-site. */
      new ConditionalPropertyFilter("hbase-site", "hbase.rpc.controllerfactory.class",
        "org.apache.hadoop.hbase.ipc.controller.ServerRpcControllerFactory"),
      new HDFSNameNodeHAFilter(),
      new HawqHAFilter() };

  private ClusterTopology clusterTopology;


  public BlueprintConfigurationProcessor(ClusterTopology clusterTopology) {
    this.clusterTopology = clusterTopology;
    initRemovePropertyUpdaters();
  }

  public Map<String, Map<String, PropertyUpdater>> getRemovePropertyUpdaters() {
    return removePropertyUpdaters;
  }

  public void initRemovePropertyUpdaters() {

    if (containsHostFromHostGroups("oozie-site", "oozie.service.JPAService.jdbc.url")) {
      Map<String, PropertyUpdater> oozieSiteUpdaters = singleHostTopologyUpdaters.get("oozie-site");
      Map<String, PropertyUpdater> oozieEnvUpdaters = singleHostTopologyUpdaters.get("oozie-env");
      if (oozieSiteUpdaters == null) {
        oozieSiteUpdaters = new HashMap<>();
      }
      if (oozieEnvUpdaters == null) {
        oozieEnvUpdaters = new HashMap<>();
      }
      oozieEnvUpdaters.put("oozie_existing_mysql_host", new SingleHostTopologyUpdater("OOZIE_SERVER"));
      oozieEnvUpdaters.put("oozie_existing_oracle_host", new SingleHostTopologyUpdater("OOZIE_SERVER"));
      oozieEnvUpdaters.put("oozie_existing_postgresql_host", new SingleHostTopologyUpdater("OOZIE_SERVER"));
      oozieSiteUpdaters.put("oozie.service.JPAService.jdbc.url",  new SingleHostTopologyUpdater("OOZIE_SERVER"));

      singleHostTopologyUpdaters.put("oozie-env", oozieEnvUpdaters);
      singleHostTopologyUpdaters.put("oozie-site", oozieSiteUpdaters);
    } else {
      Map<String, PropertyUpdater> oozieEnvOriginalValueMap = new HashMap<>();
      Map<String, PropertyUpdater> oozieSiteOriginalValueMap = new HashMap<>();
      // register updaters for Oozie properties that may point to an external DB
      oozieEnvOriginalValueMap.put("oozie_existing_mysql_host", new OriginalValuePropertyUpdater());
      oozieEnvOriginalValueMap.put("oozie_existing_oracle_host", new OriginalValuePropertyUpdater());
      oozieEnvOriginalValueMap.put("oozie_existing_postgresql_host", new OriginalValuePropertyUpdater());
      oozieSiteOriginalValueMap.put("oozie.service.JPAService.jdbc.url", new OriginalValuePropertyUpdater());

      removePropertyUpdaters.put("oozie-env", oozieEnvOriginalValueMap);
      removePropertyUpdaters.put("oozie-site", oozieSiteOriginalValueMap);
    }

    Map<String, PropertyUpdater> hiveEnvOriginalValueMap = new HashMap<>();
    // register updaters for Hive properties that may point to an external DB
    hiveEnvOriginalValueMap.put("hive_existing_oracle_host", new OriginalValuePropertyUpdater());
    hiveEnvOriginalValueMap.put("hive_existing_mssql_server_2_host", new OriginalValuePropertyUpdater());
    hiveEnvOriginalValueMap.put("hive_existing_mssql_server_host", new OriginalValuePropertyUpdater());
    hiveEnvOriginalValueMap.put("hive_existing_postgresql_host", new OriginalValuePropertyUpdater());
    hiveEnvOriginalValueMap.put("hive_existing_mysql_host", new OriginalValuePropertyUpdater());

    removePropertyUpdaters.put("hive-env", hiveEnvOriginalValueMap);

  }

  private boolean containsHostFromHostGroups(String configType, String propertyName) {
    String propertyValue = clusterTopology.getConfiguration().getPropertyValue(configType, propertyName);
    if (StringUtils.isEmpty(propertyValue)) {
      return false;
    }
    // check fir bp import
    Matcher m = HostGroup.HOSTGROUP_REGEX.matcher(propertyValue);
    if (m.find()) {
      return true;
    }

    // check for bp export
    for (HostGroupInfo groupInfo : clusterTopology.getHostGroupInfo().values()) {
      Collection<String> hosts = groupInfo.getHostNames();
      for (String host : hosts) {
        if (propertyValue.contains(host)) {
          return true;
        }
      }
    }
    return false;
  }

  public Set<String> getRequiredHostGroups() {
    Set<String> requiredHostGroups = new HashSet<>();
    Collection<Map<String, Map<String, PropertyUpdater>>> updaters = createCollectionOfUpdaters();

    // Iterate all registered updaters and collect host groups referenced by related properties and
    // extracted by the updaters
    for (Map<String, Map<String, PropertyUpdater>> updaterMap : updaters) {
      for (Map.Entry<String, Map<String, PropertyUpdater>> entry : updaterMap.entrySet()) {
        String type = entry.getKey();
        for (Map.Entry<String, PropertyUpdater> updaterEntry : entry.getValue().entrySet()) {
          String propertyName = updaterEntry.getKey();
          PropertyUpdater updater = updaterEntry.getValue();

          // cluster scoped configuration which also includes all default and BP properties
          Map<String, Map<String, String>> clusterProps = clusterTopology.getConfiguration().getFullProperties();
          Map<String, String> typeMap = clusterProps.get(type);
          if (typeMap != null && typeMap.containsKey(propertyName) && typeMap.get(propertyName) != null) {
            requiredHostGroups.addAll(updater.getRequiredHostGroups(
              propertyName, typeMap.get(propertyName), clusterProps, clusterTopology));
          }

          // host group configs
          for (HostGroupInfo groupInfo : clusterTopology.getHostGroupInfo().values()) {
            Map<String, Map<String, String>> hgConfigProps = groupInfo.getConfiguration().getProperties();
            Map<String, String> hgTypeMap = hgConfigProps.get(type);
            if (hgTypeMap != null && hgTypeMap.containsKey(propertyName)) {
              requiredHostGroups.addAll(updater.getRequiredHostGroups(
                propertyName, hgTypeMap.get(propertyName), hgConfigProps, clusterTopology));
            }
          }
        }
      }
    }

    // Iterate through all user defined properties (blueprint + cluster template only, no stack defaults) that do not
    // have a registered updater. These properties can reference host groups too which should be extracted by the default
    // updater
    Set<Pair<String, String>> propertiesWithUpdaters = getAllPropertiesWithUpdaters(updaters);

    // apply default updater on cluster config
    Map<String, Map<String, String>> userDefinedClusterProperties = clusterTopology.getConfiguration().getFullProperties(1);
    addRequiredHostgroupsByDefaultUpdater(userDefinedClusterProperties, propertiesWithUpdaters, requiredHostGroups);
    // apply default updater on hostgroup configs
    clusterTopology.getHostGroupInfo().values().stream().forEach(
      hostGroup -> {
        Configuration hostGroupConfig = hostGroup.getConfiguration();
        Map<String, Map<String, String>> hostGroupConfigProps = hostGroupConfig.getFullProperties(1);
        addRequiredHostgroupsByDefaultUpdater(hostGroupConfigProps, propertiesWithUpdaters, requiredHostGroups);
      });
    return requiredHostGroups;
  }

  /**
   * Adds required host groups based on user defined (in the blueprint or cluster template) configuration properties and
   * the default property updaters. Only those properties are considered which don't have a configured updater.
   * @param properties properties to scan for host group references
   * @param propertiesWithUpdaters properties in (configType, propertyName) which have a configured updater
   * @param hostGroupAccumulator collection to accumulate required host groups
   */
  private void addRequiredHostgroupsByDefaultUpdater(Map<String, Map<String, String>> properties,
                                                     Set<Pair<String, String>> propertiesWithUpdaters,
                                                     Set<String> hostGroupAccumulator) {
    properties.entrySet().forEach(
      configTypeEntry -> {
        String configType = configTypeEntry.getKey();
        configTypeEntry.getValue().entrySet().forEach(
          propertyEntry -> {
            String propertyName = propertyEntry.getKey();
            String oldValue = propertyEntry.getValue();
            if (!propertiesWithUpdaters.contains(Pair.of(configType, propertyName))) {
              Collection<String> requiredHostGroups =
                PropertyUpdater.defaultUpdater().getRequiredHostGroups(propertyName, oldValue, properties, clusterTopology);
              if (!requiredHostGroups.isEmpty()) {
                LOG.info("The following host groups are required by applying the default property updater on {}/{} property: {}",
                  configType, propertyName, requiredHostGroups);
              }
              hostGroupAccumulator.addAll(requiredHostGroups);
            }
          });
      });
  }


  /**
   * Update properties for cluster creation.  This involves updating topology related properties with
   * concrete topology information.
   *
   * @return Set of config type names that were updated by this update call
   */
  public Set<String> doUpdateForClusterCreate() throws ConfigurationTopologyException {
    Set<String> configTypesUpdated = new HashSet<>();
    Configuration clusterConfig = clusterTopology.getConfiguration();

    doRecommendConfigurations(clusterConfig, configTypesUpdated);

    // filter out any properties that should not be included, based on the dependencies
    // specified in the stacks, and the filters defined in this class
    doFilterPriorToClusterUpdate(clusterConfig, configTypesUpdated);

    Set<String> propertiesMoved = clusterConfig.moveProperties(HADOOP_ENV_CONFIG_TYPE_NAME, HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_HA_INITIAL_PROPERTIES);
    if (!propertiesMoved.isEmpty()) {
      configTypesUpdated.add(HADOOP_ENV_CONFIG_TYPE_NAME);
      configTypesUpdated.add(HDFS_HA_INITIAL_CONFIG_TYPE);
    }

    // this needs to be called after doFilterPriorToClusterUpdate() to ensure that the returned
    // set of properties (copy) doesn't include the removed properties.  If an updater
    // removes a property other than the property it is registered for then we will
    // have an issue as it won't be removed from the clusterProps map as it is a copy.
    Map<String, Map<String, String>> clusterProps = clusterConfig.getFullProperties();
    doGeneralPropertyUpdatesForClusterCreate(clusterConfig, clusterProps, configTypesUpdated);

    //todo: lots of hard coded HA rules included here
    if (clusterTopology.isNameNodeHAEnabled()) {
      doNameNodeHAUpdateOnClusterCreation(clusterConfig, clusterProps, configTypesUpdated);
    }

    // Explicitly set any properties that are required but not currently provided in the stack definition.
    injectDefaults(clusterConfig, configTypesUpdated, clusterTopology.getBlueprint().getServices());
    setStackToolsAndFeatures(clusterConfig, configTypesUpdated);
    addExcludedConfigProperties(clusterConfig, configTypesUpdated, clusterTopology.getBlueprint().getStack());

    trimProperties(clusterConfig, clusterTopology);

    return configTypesUpdated;
  }

  /**
   * Update Namenode HA properties during cluster creation
   * @throws ConfigurationTopologyException
   */
  private void doNameNodeHAUpdateOnClusterCreation(Configuration clusterConfig,
                                                   Map<String, Map<String, String>> clusterProps,
                                                   Set<String> configTypesUpdated) throws ConfigurationTopologyException {

    final Collection<String> nnHosts = clusterTopology.getHostAssignmentsForComponent("NAMENODE");

    // external namenodes
    if (nnHosts.isEmpty()) {
      LOG.info("NAMENODE HA is enabled but there are no NAMENODE components in the cluster. Assuming external name nodes.");
      // need to redo validation here as required information (explicit hostnames for some host groups) may have been
      // missing by the time the ClusterTopology object was validated
      try {
        new NameNodeHaValidator().validateExternalNamenodeHa(clusterTopology);
      }
      catch (InvalidTopologyException ex) {
        throw new ConfigurationTopologyException(ex.getMessage(), ex);
      }
      return;
    }

    // add "dfs.internal.nameservices" if it's not specified
    Map<String, String> hdfsSiteConfig = clusterConfig.getFullProperties().get("hdfs-site");
    String nameservices = hdfsSiteConfig.get("dfs.nameservices");
    String int_nameservices = hdfsSiteConfig.get("dfs.internal.nameservices");
    if(int_nameservices == null && nameservices != null) {
      clusterConfig.setProperty("hdfs-site", "dfs.internal.nameservices", nameservices);
    }

    // parse out the nameservices value
    String[] parsedNameServices = parseNameServices(hdfsSiteConfig);

    // if a single nameservice is configured (default HDFS HA deployment)
    if (parsedNameServices.length == 1) {
      LOG.info("Processing a single HDFS NameService, which indicates a default HDFS NameNode HA deployment");
      // if the active/standby namenodes are not specified, assign them automatically
      if (! isNameNodeHAInitialActiveNodeSet(clusterProps) && ! isNameNodeHAInitialStandbyNodeSet(clusterProps)) {
        if (nnHosts.size() == 1) { // can't be 0 as in that case was handled above
          throw new ConfigurationTopologyException("NAMENODE HA requires at least two hosts running NAMENODE but there is " +
            "only one: " + nnHosts.iterator().next());
        }

        // set the properties that configure which namenode is active,
        // and which is a standby node in this HA deployment
        Iterator<String> nnHostIterator = nnHosts.iterator();
        clusterConfig.setProperty(HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_ACTIVE_NAMENODE_PROPERTY_NAME, nnHostIterator.next());
        clusterConfig.setProperty(HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_STANDBY_NAMENODE_PROPERTY_NAME, nnHostIterator.next());

        configTypesUpdated.add(HDFS_HA_INITIAL_CONFIG_TYPE);
      }
    } else {
      if (!isPropertySet(clusterProps, HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_ACTIVE_NAMENODE_SET_PROPERTY_NAME) && !isPropertySet(clusterProps, HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_STANDBY_NAMENODE_SET_PROPERTY_NAME)) {        // multiple nameservices indicates an HDFS NameNode Federation install
        // process each nameservice to determine the active/standby nodes
        LOG.info("Processing multiple HDFS NameService instances, which indicates a NameNode Federation deployment");
        if (parsedNameServices.length > 1) {
          Set<String> activeNameNodeHostnames = new HashSet<>();
          Set<String> standbyNameNodeHostnames = new HashSet<>();

          for (String nameService : parsedNameServices) {
            List<String> hostNames = new ArrayList<>();
            String[] nameNodes = parseNameNodes(nameService, hdfsSiteConfig);
            for (String nameNode : nameNodes) {
              // use the HA rpc-address property to obtain the NameNode hostnames
              String propertyName = "dfs.namenode.rpc-address." + nameService + "." + nameNode;
              String propertyValue = hdfsSiteConfig.get(propertyName);
              if (propertyValue == null) {
                throw new ConfigurationTopologyException("NameNode HA property = " + propertyName + " is not found in the cluster config.  This indicates an error in configuration for HA/Federated clusters.  " +
                  "Please recheck the HDFS configuration and try this deployment again");
              }

              String hostName = propertyValue.split(":")[0];
              hostNames.add(hostName);
            }

            if (hostNames.size() < 2) {
              throw new ConfigurationTopologyException("NAMENODE HA for nameservice = " + nameService + " requires at least 2 hosts running NAMENODE but there are: " +
                hostNames.size() + " Hosts: " + hostNames);
            } else {
              // by default, select the active and standby namenodes for this nameservice
              // using the first two hostnames found
              // since HA is assumed, there should only be two NameNodes deployed per NameService
              activeNameNodeHostnames.add(hostNames.get(0));
              standbyNameNodeHostnames.add(hostNames.get(1));
            }
          }

          // set the properties what configure the NameNode Active/Standby status for each nameservice
          if (!activeNameNodeHostnames.isEmpty() && !standbyNameNodeHostnames.isEmpty()) {
            clusterConfig.setProperty(HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_ACTIVE_NAMENODE_SET_PROPERTY_NAME, String.join(",", activeNameNodeHostnames));
            clusterConfig.setProperty(HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_STANDBY_NAMENODE_SET_PROPERTY_NAME, String.join(",", standbyNameNodeHostnames));

            // also set the clusterID property, required for Federation installs of HDFS
            if (!isPropertySet(clusterProps, HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_HA_INITIAL_CLUSTER_ID_PROPERTY_NAME)) {
              clusterConfig.setProperty(HDFS_HA_INITIAL_CONFIG_TYPE, HDFS_HA_INITIAL_CLUSTER_ID_PROPERTY_NAME, getClusterName());
            }

            configTypesUpdated.add(HDFS_HA_INITIAL_CONFIG_TYPE);
          } else {
            LOG.warn("Error in processing the set of active/standby namenodes in this federated cluster, please check hdfs-site configuration");
          }
          // Need to configure ranger-tagsync-site properties, when Ranger-HDFS plugin is enabled
          doTagSyncSiteUpdateForNamenodeNFederationEnabledOnClusterCreation(clusterConfig, clusterProps, configTypesUpdated);
        }
      }
    }
  }

  /**
   * Update ranger-tagsync-site properties when NN Federation is enabled and Ranger-HDFS plugin is enabled
   * @throws ConfigurationTopologyException
   */
  private void doTagSyncSiteUpdateForNamenodeNFederationEnabledOnClusterCreation(Configuration clusterConfig,
                                                                    Map<String, Map<String, String>> clusterProps,
                                                                    Set<String> configTypesUpdated) throws ConfigurationTopologyException {
    Map<String, String> hdfsSiteConfig = clusterConfig.getFullProperties().get("hdfs-site");
    // parse out the nameservices value
    String[] parsedNameServices = parseNameServices(hdfsSiteConfig);
    String clusterName = getClusterName();

    // Getting configuration and properties for adding configurations to ranger-tagsync-site
    boolean isRangerHDFSPluginEnabled =  false;
    String rangerHDFSPluginServiceName = "";

    String atlasServerComponentName = "ATLAS_SERVER";
    String rangerAdminComponentName = "RANGER_ADMIN";
    String rangerTagsyncComponentName = "RANGER_TAGSYNC";
    boolean isRangerAdminToBeInstalled = (clusterTopology.getHostGroupsForComponent(rangerAdminComponentName).size() >= 1);
    boolean isRangerTagsyncToBeInstalled = (clusterTopology.getHostGroupsForComponent(rangerTagsyncComponentName).size() >= 1);
    boolean isAtlasServerToBeInstalled = (clusterTopology.getHostGroupsForComponent(atlasServerComponentName).size() >= 1);
    if (isRangerAdminToBeInstalled) {
      Map<String, String> rangerHDFSPluginProperties = clusterProps.get("ranger-hdfs-plugin-properties");
      String rangerHDFSPluginEnabledValue = rangerHDFSPluginProperties.getOrDefault("ranger-hdfs-plugin-enabled","No");
      isRangerHDFSPluginEnabled = ("yes".equalsIgnoreCase(rangerHDFSPluginEnabledValue));
      Map<String, String> rangerHDFSSecurityConfig = clusterProps.get("ranger-hdfs-security");
      rangerHDFSPluginServiceName = rangerHDFSSecurityConfig.get("ranger.plugin.hdfs.service.name");
    }

    boolean isTagsyncPropertyConfigurationRequired = ( isRangerAdminToBeInstalled && isRangerTagsyncToBeInstalled &&
                                                       isAtlasServerToBeInstalled && isRangerHDFSPluginEnabled );

    Map<String, String> coreSiteConfig = clusterProps.get("core-site");
    String fsDefaultFSValue = coreSiteConfig.get("fs.defaultFS");
    String nameServiceInFsDefaultFSConfig="";

    if (isTagsyncPropertyConfigurationRequired && "{{repo_name}}".equalsIgnoreCase(rangerHDFSPluginServiceName)) {
      rangerHDFSPluginServiceName = clusterName + "_hadoop";
    }
    // If blueprint configuration has multiple nameservices, indicating NN-Federation is enabled
    if (parsedNameServices.length > 1 && isTagsyncPropertyConfigurationRequired) {
      for (String nameService : parsedNameServices) {
        // Adding configurations for Ranger-Tagsync to map Ranger HDFS service for Atlas Tagsync
        String tagsyncNameserviceMappingProperty   = "ranger.tagsync.atlas.hdfs.instance." + clusterName + ".nameservice." + nameService + ".ranger.service";
        String updatedRangerHDFSPluginServiceName = rangerHDFSPluginServiceName + "_" + nameService;
        clusterConfig.setProperty(RANGER_TAGSYNC_SITE_CONFIG_TYPE_NAME, tagsyncNameserviceMappingProperty, updatedRangerHDFSPluginServiceName);
        try {
          URI fsDefaultFSURI = new URI(fsDefaultFSValue);
          String fsDefaultFSNameService = fsDefaultFSURI.getHost();
