/*
 * 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;

import static java.util.Collections.singletonList;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;

import javax.annotation.Nullable;

import org.apache.ambari.annotations.Experimental;
import org.apache.ambari.annotations.ExperimentalFeature;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.ServiceNotFoundException;
import org.apache.ambari.server.actionmanager.ActionManager;
import org.apache.ambari.server.actionmanager.RequestFactory;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.actionmanager.StageFactory;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorException;
import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorHelper;
import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorRequest;
import org.apache.ambari.server.api.services.stackadvisor.recommendations.RecommendationResponse;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.internal.RequestResourceFilter;
import org.apache.ambari.server.controller.internal.RequestStageContainer;
import org.apache.ambari.server.controller.utilities.KerberosChecker;
import org.apache.ambari.server.metadata.RoleCommandOrder;
import org.apache.ambari.server.orm.dao.ArtifactDAO;
import org.apache.ambari.server.orm.dao.HostDAO;
import org.apache.ambari.server.orm.dao.KerberosKeytabDAO;
import org.apache.ambari.server.orm.dao.KerberosKeytabPrincipalDAO;
import org.apache.ambari.server.orm.dao.KerberosKeytabPrincipalDAO.KeytabPrincipalFindOrCreateResult;
import org.apache.ambari.server.orm.dao.KerberosPrincipalDAO;
import org.apache.ambari.server.orm.entities.ArtifactEntity;
import org.apache.ambari.server.orm.entities.HostEntity;
import org.apache.ambari.server.orm.entities.KerberosKeytabEntity;
import org.apache.ambari.server.orm.entities.KerberosKeytabPrincipalEntity;
import org.apache.ambari.server.orm.entities.KerberosPrincipalEntity;
import org.apache.ambari.server.security.credential.Credential;
import org.apache.ambari.server.security.credential.PrincipalKeyCredential;
import org.apache.ambari.server.security.encryption.CredentialStoreService;
import org.apache.ambari.server.serveraction.ServerAction;
import org.apache.ambari.server.serveraction.kerberos.CleanupServerAction;
import org.apache.ambari.server.serveraction.kerberos.Component;
import org.apache.ambari.server.serveraction.kerberos.ConfigureAmbariIdentitiesServerAction;
import org.apache.ambari.server.serveraction.kerberos.CreateKeytabFilesServerAction;
import org.apache.ambari.server.serveraction.kerberos.CreatePrincipalsServerAction;
import org.apache.ambari.server.serveraction.kerberos.DestroyPrincipalsServerAction;
import org.apache.ambari.server.serveraction.kerberos.FinalizeKerberosServerAction;
import org.apache.ambari.server.serveraction.kerberos.KDCType;
import org.apache.ambari.server.serveraction.kerberos.KerberosAdminAuthenticationException;
import org.apache.ambari.server.serveraction.kerberos.KerberosIdentityDataFileWriter;
import org.apache.ambari.server.serveraction.kerberos.KerberosInvalidConfigurationException;
import org.apache.ambari.server.serveraction.kerberos.KerberosKDCConnectionException;
import org.apache.ambari.server.serveraction.kerberos.KerberosKDCSSLConnectionException;
import org.apache.ambari.server.serveraction.kerberos.KerberosLDAPContainerException;
import org.apache.ambari.server.serveraction.kerberos.KerberosMissingAdminCredentialsException;
import org.apache.ambari.server.serveraction.kerberos.KerberosOperationException;
import org.apache.ambari.server.serveraction.kerberos.KerberosOperationHandler;
import org.apache.ambari.server.serveraction.kerberos.KerberosOperationHandlerFactory;
import org.apache.ambari.server.serveraction.kerberos.KerberosRealmException;
import org.apache.ambari.server.serveraction.kerberos.KerberosServerAction;
import org.apache.ambari.server.serveraction.kerberos.PrepareDisableKerberosServerAction;
import org.apache.ambari.server.serveraction.kerberos.PrepareEnableKerberosServerAction;
import org.apache.ambari.server.serveraction.kerberos.PrepareKerberosIdentitiesServerAction;
import org.apache.ambari.server.serveraction.kerberos.UpdateKerberosConfigsServerAction;
import org.apache.ambari.server.serveraction.kerberos.stageutils.ResolvedKerberosKeytab;
import org.apache.ambari.server.serveraction.kerberos.stageutils.ResolvedKerberosPrincipal;
import org.apache.ambari.server.stageplanner.RoleGraph;
import org.apache.ambari.server.stageplanner.RoleGraphFactory;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ComponentInfo;
import org.apache.ambari.server.state.Config;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostState;
import org.apache.ambari.server.state.PropertyInfo;
import org.apache.ambari.server.state.SecurityType;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceInfo;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.State;
import org.apache.ambari.server.state.ValueAttributesInfo;
import org.apache.ambari.server.state.kerberos.AbstractKerberosDescriptorContainer;
import org.apache.ambari.server.state.kerberos.KerberosComponentDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosConfigurationDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosDescriptorFactory;
import org.apache.ambari.server.state.kerberos.KerberosIdentityDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosKeytabDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosPrincipalDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosPrincipalType;
import org.apache.ambari.server.state.kerberos.KerberosServiceDescriptor;
import org.apache.ambari.server.state.kerberos.VariableReplacementHelper;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostServerActionEvent;
import org.apache.ambari.server.utils.StageUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.directory.server.kerberos.shared.keytab.Keytab;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.google.inject.Injector;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;

@Singleton
public class KerberosHelperImpl implements KerberosHelper {

  public static final String BASE_LOG_DIR = "/tmp/ambari";

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

  /**
   * The set of states a component may be in, indicating that is have been previously installed on
   * the cluster.
   * <p>
   * These values are important when trying to determine the state of the cluster when adding new components
   */
  private static final Set<State> PREVIOUSLY_INSTALLED_STATES = EnumSet.of(State.INSTALLED, State.STARTED, State.DISABLED);
  public static final String CHECK_KEYTABS = "CHECK_KEYTABS";
  public static final String SET_KEYTAB = "SET_KEYTAB";
  public static final String REMOVE_KEYTAB = "REMOVE_KEYTAB";

  @Inject
  private AmbariCustomCommandExecutionHelper customCommandExecutionHelper;

  @Inject
  private AmbariManagementController ambariManagementController;

  @Inject
  private AmbariMetaInfo ambariMetaInfo;

  @Inject
  private ActionManager actionManager;

  @Inject
  private RequestFactory requestFactory;

  @Inject
  private StageFactory stageFactory;

  @Inject
  private RoleGraphFactory roleGraphFactory;

  @Inject
  private Clusters clusters;

  @Inject
  private ConfigHelper configHelper;

  @Inject
  private VariableReplacementHelper variableReplacementHelper;

  @Inject
  private Configuration configuration;

  @Inject
  private KerberosOperationHandlerFactory kerberosOperationHandlerFactory;

  @Inject
  private KerberosDescriptorFactory kerberosDescriptorFactory;

  @Inject
  private ArtifactDAO artifactDAO;

  @Inject
  private KerberosPrincipalDAO kerberosPrincipalDAO;

  @Inject
  private KerberosKeytabDAO kerberosKeytabDAO;

  @Inject
  private KerberosKeytabPrincipalDAO kerberosKeytabPrincipalDAO;

  @Inject
  private HostDAO hostDAO;

  /**
   * The injector used to create new instances of helper classes like CreatePrincipalsServerAction
   * and CreateKeytabFilesServerAction.
   */
  @Inject
  private Injector injector;

  /**
   * The secure storage facility to use to store KDC administrator credential.
   */
  @Inject
  private CredentialStoreService credentialStoreService;

  @Inject
  private StackAdvisorHelper stackAdvisorHelper;

  @Override
  public RequestStageContainer toggleKerberos(Cluster cluster, SecurityType securityType,
                                              RequestStageContainer requestStageContainer,
                                              Boolean manageIdentities)
    throws AmbariException, KerberosOperationException {

    KerberosDetails kerberosDetails = getKerberosDetails(cluster, manageIdentities);

    // Update KerberosDetails with the new security type - the current one in the cluster is the "old" value
    kerberosDetails.setSecurityType(securityType);

    if (securityType == SecurityType.KERBEROS) {
      LOG.info("Configuring Kerberos for realm {} on cluster, {}", kerberosDetails.getDefaultRealm(), cluster.getClusterName());
      requestStageContainer = handle(cluster, kerberosDetails, null, null, null, null, requestStageContainer, new EnableKerberosHandler());
    } else if (securityType == SecurityType.NONE) {
      LOG.info("Disabling Kerberos from cluster, {}", cluster.getClusterName());
      requestStageContainer = handle(cluster, kerberosDetails, null, null, null, null, requestStageContainer, new DisableKerberosHandler());
    } else {
      throw new AmbariException(String.format("Unexpected security type value: %s", securityType.name()));
    }

    return requestStageContainer;
  }

  @Override
  public RequestStageContainer executeCustomOperations(Cluster cluster, Map<String, String> requestProperties,
                                                       RequestStageContainer requestStageContainer,
                                                       Boolean manageIdentities)
    throws AmbariException, KerberosOperationException {

    if (requestProperties != null) {

      for (SupportedCustomOperation operation : SupportedCustomOperation.values()) {
        if (requestProperties.containsKey(operation.name().toLowerCase())) {
          String value = requestProperties.get(operation.name().toLowerCase());

          // The operation specific logic is kept in one place and described here
          switch (operation) {
            case REGENERATE_KEYTABS:
              if (cluster.getSecurityType() != SecurityType.KERBEROS) {
                throw new AmbariException(String.format("Custom operation %s can only be requested with the security type cluster property: %s", operation.name(), SecurityType.KERBEROS.name()));
              }

              KerberosServerAction.OperationType operationType;
              if ("true".equalsIgnoreCase(value) || "all".equalsIgnoreCase(value)) {
                operationType = KerberosServerAction.OperationType.RECREATE_ALL;
              } else if ("missing".equalsIgnoreCase(value)) {
                operationType = KerberosServerAction.OperationType.CREATE_MISSING;
              } else {
                throw new AmbariException(String.format("Unexpected directive value: %s", value));
              }

              boolean retryAllowed = false;
              if (requestProperties.containsKey(ALLOW_RETRY)) {
                String allowRetryString = requestProperties.get(ALLOW_RETRY);
                retryAllowed = Boolean.parseBoolean(allowRetryString);
              }

              Set<String> hostFilter = parseHostFilter(requestProperties);
              Map<String, Set<String>> serviceComponentFilter = parseComponentFilter(requestProperties);

              UpdateConfigurationPolicy updateConfigurationsPolicy = UpdateConfigurationPolicy.ALL;
              if(requestProperties.containsKey(DIRECTIVE_CONFIG_UPDATE_POLICY)) {
                String policyValue = requestProperties.get(DIRECTIVE_CONFIG_UPDATE_POLICY);
                updateConfigurationsPolicy = UpdateConfigurationPolicy.translate(policyValue);

                if(updateConfigurationsPolicy== null) {
                  throw new AmbariException(String.format("Unexpected comfiguration policy value: %s", policyValue));
                }
              }
              else if(requestProperties.containsKey(DIRECTIVE_IGNORE_CONFIGS)) {
                if("true".equalsIgnoreCase(requestProperties.get(DIRECTIVE_IGNORE_CONFIGS))) {
                  // This really means to no update existing properties. However, we need to ensure
                  // that Kerberos identity specific configurations are updated or added to make
                  // sure all is consistent.
                  updateConfigurationsPolicy = UpdateConfigurationPolicy.NEW_AND_IDENTITIES;
                }
              }

              boolean forceAllHosts = (hostFilter == null) || (hostFilter.contains("*"));

              CreatePrincipalsAndKeytabsHandler handler = new CreatePrincipalsAndKeytabsHandler(operationType, updateConfigurationsPolicy, forceAllHosts, true);
              handler.setRetryAllowed(retryAllowed);

              requestStageContainer = handle(cluster, getKerberosDetails(cluster, manageIdentities),
                  serviceComponentFilter, hostFilter, null, null, requestStageContainer, handler);

              break;

            default: // No other operations are currently supported
              throw new AmbariException(String.format("Custom operation not supported: %s", operation.name()));
          }
        }
      }
    }

    return requestStageContainer;
  }

  /**
   * Parsing 'Kerberos/hosts' property to get list of hosts for 'regenerate_keytabs' request.
   * Must be a string with coma separated list of hosts. Absent or miss-spelled hosts must be silently ignored
   * by caller code.
   *
   * @param requestProperties
   * @return
   */
  public static Set<String> parseHostFilter(final Map<String, String> requestProperties) {
    if (requestProperties.containsKey(DIRECTIVE_HOSTS)) {
      return ImmutableSet.copyOf(requestProperties.get(DIRECTIVE_HOSTS).split(","));
    }
    return null;
  }

  /**
   * Parsing 'Kerberos/components' property to get list of components for 'regenerate_keytabs' request.
   * Must be a comma separated list of strings that follow pattern 'SERVICENAME:COMPONENTNAME;ANOTHERCOMPONENTNAME'.
   * For example: HDFS:NAMENODE;DATANODE,YARN:RESOURCEMANAGER,ZOOKEEPER:ZOOKEEPER_SERVER;ZOOKEEPER_CLIENT.
   * Absent or miss-spelled components and services must be silently ignored by caller code.
   *
   * @param requestProperties
   * @return
   */
  public static Map<String, Set<String>> parseComponentFilter(final Map<String, String> requestProperties) {
    if (requestProperties.containsKey(DIRECTIVE_COMPONENTS)) {
      ImmutableMap.Builder<String, Set<String>> serviceComponentFilter = ImmutableMap.builder();
      for (String serviceString : requestProperties.get(DIRECTIVE_COMPONENTS).split(",")) {
        String[] serviceComponentsArray = serviceString.split(":");
        String serviceName = serviceComponentsArray[0];
        if (serviceComponentsArray.length == 2) {
          serviceComponentFilter.put(serviceName, ImmutableSet.copyOf(serviceComponentsArray[1].split(";")));
        } else {
          serviceComponentFilter.put(serviceName, ImmutableSet.of("*"));
        }
      }
      return serviceComponentFilter.build();
    }
    return null;
  }

  @Override
  public RequestStageContainer ensureIdentities(Cluster cluster, Map<String, ? extends Collection<String>> serviceComponentFilter,
                                                Set<String> hostFilter, Collection<String> identityFilter, Set<String> hostsToForceKerberosOperations,
                                                RequestStageContainer requestStageContainer, Boolean manageIdentities)
    throws AmbariException, KerberosOperationException {
    return handle(cluster,
        getKerberosDetails(cluster, manageIdentities),
        serviceComponentFilter,
        hostFilter,
        identityFilter,
        hostsToForceKerberosOperations,
        requestStageContainer,
        new CreatePrincipalsAndKeytabsHandler(KerberosServerAction.OperationType.DEFAULT,
            UpdateConfigurationPolicy.NONE,
            false,
            false));
  }

  @Override
  public RequestStageContainer deleteIdentities(Cluster cluster, Map<String, ? extends Collection<String>> serviceComponentFilter,
                                                Set<String> hostFilter, Collection<String> identityFilter, RequestStageContainer requestStageContainer,
                                                Boolean manageIdentities)
    throws AmbariException, KerberosOperationException {
    return handle(cluster, getKerberosDetails(cluster, manageIdentities), serviceComponentFilter, hostFilter, identityFilter, null,
      requestStageContainer, new DeletePrincipalsAndKeytabsHandler());
  }

  /**
   * Deletes the kerberos identities of the given component, even if the component is already deleted.
   */
  @Override
  public void deleteIdentities(Cluster cluster, List<Component> components, Set<String> identities) throws AmbariException, KerberosOperationException {
    LOG.info("Deleting identities: ", identities);
    KerberosDetails kerberosDetails = getKerberosDetails(cluster, null);
    validateKDCCredentials(kerberosDetails, cluster);
    File dataDirectory = createTemporaryDirectory();
    RoleCommandOrder roleCommandOrder = ambariManagementController.getRoleCommandOrder(cluster);
    DeleteIdentityHandler handler = new DeleteIdentityHandler(customCommandExecutionHelper, configuration.getDefaultServerTaskTimeout(), stageFactory, ambariManagementController);
    DeleteIdentityHandler.CommandParams commandParameters = new DeleteIdentityHandler.CommandParams(
      components,
      identities,
      ambariManagementController.getAuthName(),
      dataDirectory,
      kerberosDetails.getDefaultRealm(),
      kerberosDetails.getKdcType());
    OrderedRequestStageContainer stageContainer = new OrderedRequestStageContainer(
      roleGraphFactory,
      roleCommandOrder,
      new RequestStageContainer(actionManager.getNextRequestId(), null, requestFactory, actionManager));
    handler.addDeleteIdentityStages(cluster, stageContainer, commandParameters, kerberosDetails.manageIdentities());
    stageContainer.getRequestStageContainer().persist();
  }

  @Override
  public void configureServices(Cluster cluster, Map<String, Collection<String>> serviceFilter)
    throws AmbariException, KerberosInvalidConfigurationException {
    final Map<String, Set<String>> installedServices = new HashMap<>();
    final Set<String> previouslyExistingServices = new HashSet<>();

    // Calculate the map of installed services to installed components.
    // We can create the map in the "shouldIncludeCommand" Command to avoid having to iterate
    // over the returned ServiceComponentHost List.
    getServiceComponentHosts(cluster,
      sch -> {
        if (sch != null) {
          String serviceName = sch.getServiceName();

          Set<String> installedComponents = installedServices.computeIfAbsent(serviceName, k -> new HashSet<>());
          installedComponents.add(sch.getServiceComponentName());

          // Determine if this component was PREVIOUSLY installed, which implies that its containing service was PREVIOUSLY installed
          if (!previouslyExistingServices.contains(serviceName) && PREVIOUSLY_INSTALLED_STATES.contains(sch.getState())) {
            previouslyExistingServices.add(serviceName);
          }

          return true;
        }

        return false;
      });

    Map<String, Map<String, String>> existingConfigurations = configHelper.calculateExistingConfigurations(ambariManagementController, cluster, null, null);
    Map<String, Map<String, String>> updates = getServiceConfigurationUpdates(cluster,
      existingConfigurations, installedServices, serviceFilter, previouslyExistingServices, true, true);

    // Store the updates...
    for (Map.Entry<String, Map<String, String>> entry : updates.entrySet()) {
      configHelper.updateConfigType(cluster, cluster.getDesiredStackVersion(), ambariManagementController,
              entry.getKey(), entry.getValue(), null, ambariManagementController.getAuthName(),
              "Enabling Kerberos for added components");
    }
  }

  @Override
  public Map<String, Map<String, String>> getServiceConfigurationUpdates(Cluster cluster,
                                                                         Map<String, Map<String, String>> existingConfigurations,
                                                                         Map<String, Set<String>> installedServices,
                                                                         Map<String, Collection<String>> serviceFilter,
                                                                         Set<String> previouslyExistingServices,
                                                                         boolean kerberosEnabled,
                                                                         boolean applyStackAdvisorUpdates)
    throws KerberosInvalidConfigurationException, AmbariException {

    Map<String, Map<String, String>> kerberosConfigurations = new HashMap<>();
    KerberosDetails kerberosDetails = getKerberosDetails(cluster, null);
    KerberosDescriptor kerberosDescriptor = getKerberosDescriptor(cluster, false);

    Map<String, String> kerberosDescriptorProperties = kerberosDescriptor.getProperties();
    Map<String, Map<String, String>> configurations = addAdditionalConfigurations(cluster,
      deepCopy(existingConfigurations), null, kerberosDescriptorProperties, null);

    Map<String, Set<String>> propertiesToIgnore = new HashMap<>();

    // If Ambari is managing it own identities then add AMBARI to the set of installed service so
    // that its Kerberos descriptor entries will be included.
    if (createAmbariIdentities(existingConfigurations.get(KERBEROS_ENV))) {
      installedServices = new HashMap<>(installedServices);
      installedServices.put(RootService.AMBARI.name(), Collections.singleton(RootComponent.AMBARI_SERVER.name()));
    }

    // Create the context to use for filtering Kerberos Identities based on the state of the cluster
    Map<String, Object> filterContext = new HashMap<>();
    filterContext.put("configurations", configurations);
    filterContext.put("services", installedServices.keySet());

    for (Map.Entry<String, Set<String>> installedServiceEntry : installedServices.entrySet()) {
      String installedService = installedServiceEntry.getKey();

      if ((serviceFilter == null) || (serviceFilter.containsKey(installedService))) {
        Collection<String> componentFilter = (serviceFilter == null) ? null : serviceFilter.get(installedService);
        Set<String> installedComponents = installedServiceEntry.getValue();

        // Set properties...
        KerberosServiceDescriptor serviceDescriptor = kerberosDescriptor.getService(installedService);

        if (serviceDescriptor != null) {
          if (installedComponents != null) {
            boolean servicePreviouslyExisted = (previouslyExistingServices != null) && previouslyExistingServices.contains(installedService);

            for (String installedComponent : installedComponents) {

              if ((componentFilter == null) || componentFilter.contains(installedComponent)) {
                KerberosComponentDescriptor componentDescriptor = serviceDescriptor.getComponent(installedComponent);
                if (componentDescriptor != null) {
                  Map<String, Map<String, String>> identityConfigurations;

                  identityConfigurations = getIdentityConfigurations(serviceDescriptor.getIdentities(true, filterContext));
                  processIdentityConfigurations(identityConfigurations, kerberosConfigurations, configurations, propertiesToIgnore);

                  identityConfigurations = getIdentityConfigurations(componentDescriptor.getIdentities(true, filterContext));
                  processIdentityConfigurations(identityConfigurations, kerberosConfigurations, configurations, propertiesToIgnore);

                  mergeConfigurations(kerberosConfigurations,
                    componentDescriptor.getConfigurations(!servicePreviouslyExisted), configurations, null);
                }
              }
            }
          }
        }
      }
    }

    setAuthToLocalRules(cluster, kerberosDescriptor, kerberosDetails.getDefaultRealm(), installedServices, configurations, kerberosConfigurations, false);

    return (applyStackAdvisorUpdates)
      ? applyStackAdvisorUpdates(cluster, installedServices.keySet(), configurations, kerberosConfigurations, propertiesToIgnore,
      new HashMap<>(), kerberosEnabled)
      : kerberosConfigurations;
  }

  /**
   * Adds host assignments, recommended by the Stack Advisor, to the configuration map (clusterHostInfo)
   * for the components specified in the component filter <code>componentFilter</code> (or all if the
   * component filter is <code>null</code>).
   *
   * @param cluster         the cluster
   * @param services        the relevant services to consider
   * @param componentFilter the set of components to add to the clusterHostInfo structure
   * @param configurations  the configurations map to update
   * @throws AmbariException if an error occurs
   */
  private void applyStackAdvisorHostRecommendations(Cluster cluster,
                                                    Set<String> services,
                                                    Set<String> componentFilter,
                                                    Map<String, Map<String, String>> configurations)
    throws AmbariException {
    StackId stackVersion = cluster.getCurrentStackVersion();
    List<String> hostNames = new ArrayList<>();

    Collection<Host> hosts = cluster.getHosts();
    if (hosts != null) {
      for (Host host : hosts) {
        hostNames.add(host.getHostName());
      }
    }

    StackAdvisorRequest request = StackAdvisorRequest.StackAdvisorRequestBuilder
      .forStack(stackVersion.getStackName(), stackVersion.getStackVersion())
      .forServices(services)
      .forHosts(hostNames)
      .withComponentHostsMap(cluster.getServiceComponentHostMap(null, services))
      .ofType(StackAdvisorRequest.StackAdvisorRequestType.HOST_GROUPS)
      .build();

    try {
      RecommendationResponse response = stackAdvisorHelper.recommend(request);

      RecommendationResponse.Recommendation recommendation = (response == null) ? null : response.getRecommendations();
      RecommendationResponse.Blueprint blueprint = (recommendation == null) ? null : recommendation.getBlueprint();
      Set<RecommendationResponse.HostGroup> hostGroups = (blueprint == null) ? null : blueprint.getHostGroups();


      if (hostGroups != null) {
        RecommendationResponse.BlueprintClusterBinding blueprintBinding = recommendation.getBlueprintClusterBinding();
        Map<String, RecommendationResponse.BindingHostGroup> bindingMap = new HashMap<>();
        if (blueprintBinding != null) {
          Set<RecommendationResponse.BindingHostGroup> bindingHostGroups = blueprintBinding.getHostGroups();
          if (bindingHostGroups != null) {
            for (RecommendationResponse.BindingHostGroup bindingHostGroup : bindingHostGroups) {
              bindingMap.put(bindingHostGroup.getName(), bindingHostGroup);
            }
          }
        }

        // Get (and create if needed) the clusterHostInfo map
        Map<String, String> clusterHostInfoMap = configurations.computeIfAbsent(CLUSTER_HOST_INFO, __ -> new HashMap<>());

        // Iterate through the recommendations to find the recommended host assignments
        for (RecommendationResponse.HostGroup hostGroup : hostGroups) {
          Set<Map<String, String>> components = hostGroup.getComponents();

          if (components != null) {
            RecommendationResponse.BindingHostGroup binding = bindingMap.get(hostGroup.getName());

            if (binding != null) {
              Set<Map<String, String>> hostGroupHosts = binding.getHosts();

              if (hostGroupHosts != null) {
                for (Map<String, String> component : components) {
                  String componentName = component.get("name");

                  // If the component filter is null or the current component is found in the filter,
                  // include it in the map
                  if ((componentFilter == null) || componentFilter.contains(componentName)) {
                    String key = StageUtils.getClusterHostInfoKey(componentName);

                    Set<String> fqdns = new TreeSet<>();

                    // Values are a comma-delimited list of hosts.
                    // If a value exists, split it and add the tokens to the set
                    if (!StringUtils.isEmpty(clusterHostInfoMap.get(key))) {
                      fqdns.addAll(Arrays.asList(clusterHostInfoMap.get(key).split(",")));
                    }

                    // Add the set of hosts for the current host group
                    for (Map<String, String> hostGroupHost : hostGroupHosts) {
                      String fqdn = hostGroupHost.get("fqdn");

                      if (!StringUtils.isEmpty(fqdn)) {
                        fqdns.add(fqdn);
                      }
                    }

                    // create the comma-delimited list of hosts
                    clusterHostInfoMap.put(key, StringUtils.join(fqdns, ','));
                  }
                }
              }
            }
          }
        }
      }
    } catch (StackAdvisorException e) {
      LOG.error("Failed to obtain the recommended host groups for the preconfigured components.", e);
      throw new AmbariException(e.getMessage(), e);
    }
  }

  @Override
  public Map<String, Map<String, String>> applyStackAdvisorUpdates(Cluster cluster, Set<String> services,
                                                                   Map<String, Map<String, String>> existingConfigurations,
                                                                   Map<String, Map<String, String>> kerberosConfigurations,
