/*
 * 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 org.apache.ambari.server.agent.ExecutionCommand.KeyNames.AGENT_STACK_RETRY_COUNT;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.AGENT_STACK_RETRY_ON_UNAVAILABILITY;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.CLIENTS_TO_UPDATE_CONFIGS;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.COMMAND_TIMEOUT;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.COMPONENT_CATEGORY;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.CUSTOM_COMMAND;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.DB_DRIVER_FILENAME;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.DB_NAME;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.GPL_LICENSE_ACCEPTED;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.GROUP_LIST;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.HOST_SYS_PREPPED;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.JDK_LOCATION;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.MYSQL_JDBC_URL;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.NOT_MANAGED_HDFS_PATH_LIST;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.ORACLE_JDBC_URL;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.SCRIPT;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.SCRIPT_TYPE;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.STACK_NAME;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.STACK_VERSION;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.USER_GROUPS;
import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.USER_LIST;
import static org.apache.ambari.server.controller.internal.RequestResourceProvider.HAS_RESOURCE_FILTERS;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.TreeMap;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.agent.AgentCommand.AgentCommandType;
import org.apache.ambari.server.agent.ExecutionCommand;
import org.apache.ambari.server.agent.ExecutionCommand.KeyNames;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.internal.RequestOperationLevel;
import org.apache.ambari.server.controller.internal.RequestResourceFilter;
import org.apache.ambari.server.controller.internal.RequestResourceProvider;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.metadata.ActionMetadata;
import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.CommandScriptDefinition;
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.CustomCommandDefinition;
import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostComponentAdminState;
import org.apache.ambari.server.state.HostState;
import org.apache.ambari.server.state.MaintenanceState;
import org.apache.ambari.server.state.PropertyInfo;
import org.apache.ambari.server.state.PropertyInfo.PropertyType;
import org.apache.ambari.server.state.RefreshCommandConfiguration;
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.StackInfo;
import org.apache.ambari.server.state.State;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpInProgressEvent;
import org.apache.ambari.server.utils.StageUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableMap;
import com.google.gson.Gson;
import com.google.inject.Inject;
import com.google.inject.Singleton;

/**
 * Helper class containing logic to process custom command execution requests .
 * This class has special support needed for SERVICE_CHECK and DECOMMISSION.
 * These commands are not pass through as Ambari has specific persistence requirements.
 */
@Singleton
public class AmbariCustomCommandExecutionHelper {
  private final static Logger LOG = LoggerFactory.getLogger(
      AmbariCustomCommandExecutionHelper.class);

  // TODO: Remove the hard-coded mapping when stack definition indicates which slave types can be decommissioned
  static final Map<String, String> masterToSlaveMappingForDecom = ImmutableMap.<String, String>builder()
    .put("NAMENODE", "DATANODE")
    .put("RESOURCEMANAGER", "NODEMANAGER")
    .put("HBASE_MASTER", "HBASE_REGIONSERVER")
    .put("JOBTRACKER", "TASKTRACKER")
    .build();

  public final static String DECOM_INCLUDED_HOSTS = "included_hosts";
  public final static String DECOM_EXCLUDED_HOSTS = "excluded_hosts";
  public final static String ALL_DECOMMISSIONED_HOSTS = "all_decommissioned_hosts";
  public final static String DECOM_SLAVE_COMPONENT = "slave_type";
  public final static String HBASE_MARK_DRAINING_ONLY = "mark_draining_only";
  public final static String UPDATE_FILES_ONLY = "update_files_only";
  public final static String IS_ADD_OR_DELETE_SLAVE_REQUEST = "is_add_or_delete_slave_request";

  private final static String ALIGN_MAINTENANCE_STATE = "align_maintenance_state";

  public final static int MIN_STRICT_SERVICE_CHECK_TIMEOUT = 120;

  @Inject
  private ActionMetadata actionMetadata;

  @Inject
  private Clusters clusters;

  @Inject
  private AmbariManagementController managementController;

  @Inject
  private Gson gson;

  @Inject
  private Configuration configs;

  @Inject
  private AmbariMetaInfo ambariMetaInfo;

  @Inject
  private ConfigHelper configHelper;

  @Inject
  private MaintenanceStateHelper maintenanceStateHelper;

  @Inject
  private HostRoleCommandDAO hostRoleCommandDAO;

  private Map<String, Map<String, Map<String, String>>> configCredentialsForService = new HashMap<>();

  protected static final String SERVICE_CHECK_COMMAND_NAME = "SERVICE_CHECK";
  protected static final String START_COMMAND_NAME = "START";
  protected static final String RESTART_COMMAND_NAME = "RESTART";
  protected static final String INSTALL_COMMAND_NAME = "INSTALL";
  public static final String DECOMMISSION_COMMAND_NAME = "DECOMMISSION";


  private Boolean isServiceCheckCommand(String command, String service) {
    List<String> actions = actionMetadata.getActions(service);

    return !(actions == null || actions.size() == 0) && actions.contains(command);
  }

  private Boolean isValidCustomCommand(String clusterName,
      String serviceName, String componentName, String commandName)
      throws AmbariException {

    if (componentName == null) {
      return false;
    }

    Cluster cluster = clusters.getCluster(clusterName);
    Service service = cluster.getService(serviceName);
    ServiceComponent component = service.getServiceComponent(componentName);
    StackId stackId = component.getDesiredStackId();

    ComponentInfo componentInfo = ambariMetaInfo.getComponent(
        stackId.getStackName(), stackId.getStackVersion(),
        serviceName, componentName);

    return !(!componentInfo.isCustomCommand(commandName) &&
      !actionMetadata.isDefaultHostComponentCommand(commandName));
  }

  private Boolean isValidCustomCommand(ActionExecutionContext
      actionExecutionContext, RequestResourceFilter resourceFilter)
      throws AmbariException {

    if (actionExecutionContext.isFutureCommand()) {
      return true;
    }

    String clusterName = actionExecutionContext.getClusterName();
    String serviceName = resourceFilter.getServiceName();
    String componentName = resourceFilter.getComponentName();
    String commandName = actionExecutionContext.getActionName();

    if (componentName == null) {
      return false;
    }

    return isValidCustomCommand(clusterName, serviceName, componentName, commandName);
  }

  private Boolean isValidCustomCommand(ExecuteActionRequest actionRequest,
      RequestResourceFilter resourceFilter) throws AmbariException {
    String clusterName = actionRequest.getClusterName();
    String serviceName = resourceFilter.getServiceName();
    String componentName = resourceFilter.getComponentName();
    String commandName = actionRequest.getCommandName();

    if (componentName == null) {
      return false;
    }

    return isValidCustomCommand(clusterName, serviceName, componentName, commandName);
  }

  private String getReadableCustomCommandDetail(ActionExecutionContext
        actionRequest, RequestResourceFilter resourceFilter) {
    StringBuilder sb = new StringBuilder();
    sb.append(actionRequest.getActionName());
    if (resourceFilter.getServiceName() != null
        && !resourceFilter.getServiceName().equals("")) {
      sb.append(" ");
      sb.append(resourceFilter.getServiceName());
    }

    if (resourceFilter.getComponentName() != null
        && !resourceFilter.getComponentName().equals("")) {
      sb.append("/");
      sb.append(resourceFilter.getComponentName());
    }

    return sb.toString();
  }

  /**
   * Called during the start/stop/restart of services, plus custom commands during Stack Upgrade.
   * @param actionExecutionContext Execution Context
   * @param resourceFilter Resource Filter
   * @param stage Command stage
   * @param additionalCommandParams Additional command params to add the the stage
   * @param commandDetail String for the command detail
   * @throws AmbariException
   */
  private void addCustomCommandAction(final ActionExecutionContext actionExecutionContext,
      final RequestResourceFilter resourceFilter, Stage stage, Map<String, String> additionalCommandParams,
      String commandDetail, Map<String, String> requestParams) throws AmbariException {
    final String serviceName = resourceFilter.getServiceName();
    final String componentName = resourceFilter.getComponentName();
    final String commandName = actionExecutionContext.getActionName();
    boolean retryAllowed = actionExecutionContext.isRetryAllowed();
    boolean autoSkipFailure = actionExecutionContext.isFailureAutoSkipped();

    String clusterName = stage.getClusterName();
    final Cluster cluster = clusters.getCluster(clusterName);

    // start with all hosts
    Set<String> candidateHosts = new HashSet<>(resourceFilter.getHostNames());

    // Filter hosts that are in MS
    Set<String> ignoredHosts = maintenanceStateHelper.filterHostsInMaintenanceState(
      candidateHosts, new MaintenanceStateHelper.HostPredicate() {
        @Override
        public boolean shouldHostBeRemoved(final String hostname)
            throws AmbariException {
          if (actionExecutionContext.isFutureCommand()) {
            return false;
          }

          return !maintenanceStateHelper.isOperationAllowed(
              cluster, actionExecutionContext.getOperationLevel(),
              resourceFilter, serviceName, componentName, hostname);
        }
      }
    );

    // Filter unhealthy hosts
    Set<String> unhealthyHosts = getUnhealthyHosts(candidateHosts, actionExecutionContext, resourceFilter);

    // log excluded hosts
    if (!ignoredHosts.isEmpty()) {
      if( LOG.isDebugEnabled() ){
        LOG.debug(
            "While building the {} custom command for {}/{}, the following hosts were excluded: unhealthy[{}], maintenance[{}]",
            commandName, serviceName, componentName, StringUtils.join(unhealthyHosts, ','),
            StringUtils.join(ignoredHosts, ','));
      }
    } else if (!unhealthyHosts.isEmpty()) {
      if (LOG.isDebugEnabled()) {
        LOG.debug(
            "While building the {} custom command for {}/{}, the following hosts were excluded: unhealthy[{}], maintenance[{}]",
            commandName, serviceName, componentName, StringUtils.join(unhealthyHosts, ','),
            StringUtils.join(ignoredHosts, ','));
      }
    } else if (candidateHosts.isEmpty()) {
      String message = MessageFormat.format(
          "While building the {0} custom command for {1}/{2}, there were no healthy eligible hosts",
          commandName, serviceName, componentName);

      throw new AmbariException(message);
    }

    Service service = cluster.getService(serviceName);

    // grab the stack ID from the service first, and use the context's if it's set
    StackId stackId = service.getDesiredStackId();
    if (null != actionExecutionContext.getStackId()) {
      stackId = actionExecutionContext.getStackId();
    }

    AmbariMetaInfo ambariMetaInfo = managementController.getAmbariMetaInfo();
    ServiceInfo serviceInfo = ambariMetaInfo.getService(service);

    CustomCommandDefinition customCommandDefinition = null;
    ComponentInfo ci = serviceInfo.getComponentByName(componentName);
    if(ci != null){
      customCommandDefinition = ci.getCustomCommandByName(commandName);
    }

    long nowTimestamp = System.currentTimeMillis();

    for (String hostName : candidateHosts) {
      stage.addHostRoleExecutionCommand(hostName, Role.valueOf(componentName),
          RoleCommand.CUSTOM_COMMAND,
          new ServiceComponentHostOpInProgressEvent(componentName, hostName, nowTimestamp),
          cluster.getClusterName(), serviceName, retryAllowed, autoSkipFailure);

      ExecutionCommand execCmd = stage.getExecutionCommandWrapper(hostName,
          componentName).getExecutionCommand();

      // if the command should fetch brand new configuration tags before
      // execution, then we don't need to fetch them now
      if(actionExecutionContext.getParameters() != null && actionExecutionContext.getParameters().containsKey(KeyNames.OVERRIDE_CONFIGS)){
        execCmd.setOverrideConfigs(true);
      }

      HostRoleCommand cmd = stage.getHostRoleCommand(hostName, componentName);
      if (cmd != null) {
        cmd.setCommandDetail(commandDetail);
        cmd.setCustomCommandName(commandName);
        if (customCommandDefinition != null){
          cmd.setOpsDisplayName(customCommandDefinition.getOpsDisplayName());
        }
      }

      //set type background
      if(customCommandDefinition != null && customCommandDefinition.isBackground()){
        cmd.setBackgroundCommand(true);
        execCmd.setCommandType(AgentCommandType.BACKGROUND_EXECUTION_COMMAND);
      }

      execCmd.setComponentVersions(cluster);

      execCmd.setConfigurations(new TreeMap<>());

      // Get the value of credential store enabled from the DB
      Service clusterService = cluster.getService(serviceName);
      execCmd.setCredentialStoreEnabled(String.valueOf(clusterService.isCredentialStoreEnabled()));

      // Get the map of service config type to password properties for the service
      Map<String, Map<String, String>> configCredentials;
      configCredentials = configCredentialsForService.get(clusterService.getName());
      if (configCredentials == null) {
        configCredentials = configHelper.getCredentialStoreEnabledProperties(stackId, clusterService);
        configCredentialsForService.put(clusterService.getName(), configCredentials);
      }

      execCmd.setConfigurationCredentials(configCredentials);

      Map<String, String> hostLevelParams = new TreeMap<>();
      hostLevelParams.put(STACK_NAME, stackId.getStackName());
      hostLevelParams.put(STACK_VERSION, stackId.getStackVersion());

      Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs();

      Set<String> userSet = configHelper.getPropertyValuesWithPropertyType(stackId, PropertyType.USER, cluster, desiredConfigs);
      String userList = gson.toJson(userSet);
      hostLevelParams.put(USER_LIST, userList);

      //Create a user_group mapping and send it as part of the hostLevelParams
      Map<String, Set<String>> userGroupsMap = configHelper.createUserGroupsMap(
        stackId, cluster, desiredConfigs);
      String userGroups = gson.toJson(userGroupsMap);
      hostLevelParams.put(USER_GROUPS, userGroups);

      Set<String> groupSet = configHelper.getPropertyValuesWithPropertyType(stackId, PropertyType.GROUP, cluster, desiredConfigs);
      String groupList = gson.toJson(groupSet);
      hostLevelParams.put(GROUP_LIST, groupList);

      Map<PropertyInfo, String> notManagedHdfsPathMap = configHelper.getPropertiesWithPropertyType(stackId, PropertyType.NOT_MANAGED_HDFS_PATH, cluster, desiredConfigs);
      Set<String> notManagedHdfsPathSet = configHelper.filterInvalidPropertyValues(notManagedHdfsPathMap, NOT_MANAGED_HDFS_PATH_LIST);
      String notManagedHdfsPathList = gson.toJson(notManagedHdfsPathSet);
      hostLevelParams.put(NOT_MANAGED_HDFS_PATH_LIST, notManagedHdfsPathList);

      execCmd.setHostLevelParams(hostLevelParams);

      Map<String, String> commandParams = new TreeMap<>();
      if (additionalCommandParams != null) {
        for (String key : additionalCommandParams.keySet()) {
          commandParams.put(key, additionalCommandParams.get(key));
        }
      }
      commandParams.put(CUSTOM_COMMAND, commandName);

      boolean isInstallCommand = commandName.equals(RoleCommand.INSTALL.toString());
      int commandTimeout = Integer.valueOf(configs.getDefaultAgentTaskTimeout(isInstallCommand));

      ComponentInfo componentInfo = ambariMetaInfo.getComponent(
          stackId.getStackName(), stackId.getStackVersion(),
          serviceName, componentName);

      if (serviceInfo.getSchemaVersion().equals(AmbariMetaInfo.SCHEMA_VERSION_2)) {
        // Service check command is not custom command
        CommandScriptDefinition script = componentInfo.getCommandScript();

        if (script != null) {
          commandParams.put(SCRIPT, script.getScript());
          commandParams.put(SCRIPT_TYPE, script.getScriptType().toString());
          if (script.getTimeout() > 0) {
            commandTimeout = script.getTimeout();
          }
        } else {
          String message = String.format("Component %s has not command script " +
              "defined. It is not possible to send command for " +
              "this service", componentName);
          throw new AmbariException(message);
        }
        // We don't need package/repo information to perform service check
      }

      // !!! the action execution context timeout is the final say, but make sure it's at least 60 seconds
      if (null != actionExecutionContext.getTimeout()) {
        commandTimeout = actionExecutionContext.getTimeout().intValue();
        commandTimeout = Math.max(60, commandTimeout);
      }

      if (requestParams != null && requestParams.containsKey(RequestResourceProvider.CONTEXT)) {
        String requestContext = requestParams.get(RequestResourceProvider.CONTEXT);
        if (StringUtils.isNotEmpty(requestContext) && requestContext.toLowerCase().contains("rolling-restart")) {
          Config clusterEnvConfig = cluster.getDesiredConfigByType("cluster-env");
          if (clusterEnvConfig != null) {
            String componentRollingRestartTimeout = clusterEnvConfig.getProperties().get("namenode_rolling_restart_timeout");
            if (StringUtils.isNotEmpty(componentRollingRestartTimeout)) {
              commandTimeout = Integer.parseInt(componentRollingRestartTimeout);
            }
          }
        }
      }

      commandParams.put(COMMAND_TIMEOUT, "" + commandTimeout);

      Map<String, String> roleParams = execCmd.getRoleParams();
      if (roleParams == null) {
        roleParams = new TreeMap<>();
      }

      // if there is a stack upgrade which is currently suspended then pass that
      // information down with the command as some components may need to know
      boolean isUpgradeSuspended = cluster.isUpgradeSuspended();
      if (isUpgradeSuspended) {
        cluster.addSuspendedUpgradeParameters(commandParams, roleParams);
      }
      StageUtils.useAmbariJdkInCommandParams(commandParams, configs);
      roleParams.put(COMPONENT_CATEGORY, componentInfo.getCategory());

      // set reconfigureAction in case of a RECONFIGURE command if there are any
      if (commandName.equals("RECONFIGURE")) {
        String refreshConfigsCommand = configHelper.getRefreshConfigsCommand(cluster, hostName, serviceName, componentName);
        if (refreshConfigsCommand != null && !refreshConfigsCommand.equals(RefreshCommandConfiguration.REFRESH_CONFIGS)) {
              LOG.info("Refreshing configs for {}/{} with command: ", componentName, hostName, refreshConfigsCommand);
          commandParams.put("reconfigureAction", refreshConfigsCommand);
          //execCmd.setForceRefreshConfigTagsBeforeExecution(true);
        }
      }

      execCmd.setCommandParams(commandParams);
      execCmd.setRoleParams(roleParams);

      // skip anything else
      if (actionExecutionContext.isFutureCommand()) {
        continue;
      }

      // perform any server side command related logic - eg - set desired states on restart
      applyCustomCommandBackendLogic(cluster, serviceName, componentName, commandName, hostName);
    }
  }

  private void applyCustomCommandBackendLogic(Cluster cluster, String serviceName, String componentName, String commandName, String hostname) throws AmbariException {
    switch (commandName) {
      case "RESTART":
        ServiceComponent serviceComponent = cluster.getService(serviceName).getServiceComponent(componentName);
        ServiceComponentHost serviceComponentHost = serviceComponent.getServiceComponentHost(hostname);
        State currentDesiredState = serviceComponentHost.getDesiredState();

        if( !serviceComponent.isClientComponent()) {
          if (currentDesiredState != State.STARTED) {
            LOG.info("Updating desired state to {} on RESTART for {}/{} because it was {}",
                State.STARTED, serviceName, componentName, currentDesiredState);

            serviceComponentHost.setDesiredState(State.STARTED);
          }
        } else {
          LOG.debug("Desired state for client components should not be updated on RESTART. Service/Component {}/{}",
              serviceName, componentName);
        }

        break;
      default:
        LOG.debug("No backend operations needed for the custom command: {}", commandName);
        break;
    }
  }

  private void findHostAndAddServiceCheckAction(final ActionExecutionContext actionExecutionContext,
      final RequestResourceFilter resourceFilter, Stage stage) throws AmbariException {

    String clusterName = actionExecutionContext.getClusterName();
    final Cluster cluster = clusters.getCluster(clusterName);
    final String componentName = actionMetadata.getClient(resourceFilter.getServiceName());
    final String serviceName = resourceFilter.getServiceName();
    String smokeTestRole = actionMetadata.getServiceCheckAction(serviceName);
    if (null == smokeTestRole) {
      smokeTestRole = actionExecutionContext.getActionName();
    }

    Set<String> candidateHosts;
    final Map<String, ServiceComponentHost> serviceHostComponents;

    if (componentName != null) {
      serviceHostComponents = cluster.getService(serviceName).getServiceComponent(componentName).getServiceComponentHosts();

      if (serviceHostComponents.isEmpty()) {
        throw new AmbariException(MessageFormat.format("No hosts found for service: {0}, component: {1} in cluster: {2}",
            serviceName, componentName, clusterName));
      }

      // If specified a specific host, run on it as long as it contains the component.
      // Otherwise, use candidates that contain the component.
      List<String> candidateHostsList = resourceFilter.getHostNames();
      if (candidateHostsList != null && !candidateHostsList.isEmpty()) {
        candidateHosts = new HashSet<>(candidateHostsList);

        // Get the intersection.
        candidateHosts.retainAll(serviceHostComponents.keySet());

        if (candidateHosts.isEmpty()) {
          throw new AmbariException(MessageFormat.format("The resource filter for hosts does not contain components for " +
                  "service: {0}, component: {1} in cluster: {2}", serviceName, componentName, clusterName));
        }
      } else {
        candidateHosts = serviceHostComponents.keySet();
      }
    } else {
      // TODO: This code branch looks unreliable (taking random component, should prefer the clients)
      Map<String, ServiceComponent> serviceComponents = cluster.getService(serviceName).getServiceComponents();

      // Filter components without any HOST
      Iterator<String> serviceComponentNameIterator = serviceComponents.keySet().iterator();
      while (serviceComponentNameIterator.hasNext()){
        String componentToCheck = serviceComponentNameIterator.next();
         if (serviceComponents.get(componentToCheck).getServiceComponentHosts().isEmpty()){
           serviceComponentNameIterator.remove();
         }
      }

      if (serviceComponents.isEmpty()) {
        throw new AmbariException(MessageFormat.format("Did not find any hosts with components for service: {0} in cluster: {1}",
            serviceName, clusterName));
      }

      // Pick a random service (should prefer clients).
      ServiceComponent serviceComponent = serviceComponents.values().iterator().next();
      serviceHostComponents = serviceComponent.getServiceComponentHosts();
      candidateHosts = serviceHostComponents.keySet();
    }

    // check if all hostnames are valid.
    for(String candidateHostName: candidateHosts) {
      ServiceComponentHost serviceComponentHost = serviceHostComponents.get(candidateHostName);

      if (serviceComponentHost == null) {
        throw new AmbariException("Provided hostname = "
            + candidateHostName + " is either not a valid cluster host or does not satisfy the filter condition.");
      }
    }

    // Filter out hosts that are in maintenance mode - they should never be included in service checks
    Set<String> hostsInMaintenanceMode = new HashSet<>();
    if (actionExecutionContext.isMaintenanceModeHostExcluded()) {
      Iterator<String> iterator = candidateHosts.iterator();
      while (iterator.hasNext()) {
        String candidateHostName = iterator.next();
        ServiceComponentHost serviceComponentHost = serviceHostComponents.get(candidateHostName);
        Host host = serviceComponentHost.getHost();
        if (host.getMaintenanceState(cluster.getClusterId()) == MaintenanceState.ON) {
          hostsInMaintenanceMode.add(candidateHostName);
          iterator.remove();
        }
      }
    }

    // Filter out hosts that are not healthy, i.e., all hosts should be heartbeating.
    // Pick one randomly. If there are none, throw an exception.
    List<String> healthyHostNames = managementController.selectHealthyHosts(candidateHosts);
    if (healthyHostNames.isEmpty()) {
      String message = MessageFormat.format(
          "While building a service check command for {0}, there were no healthy eligible hosts: unhealthy[{1}], maintenance[{2}]",
          serviceName, StringUtils.join(candidateHosts, ','),
          StringUtils.join(hostsInMaintenanceMode, ','));

      throw new AmbariException(message);
    }

    String preferredHostName = selectRandomHostNameWithPreferenceOnAvailability(healthyHostNames);

    long nowTimestamp = System.currentTimeMillis();
    Map<String, String> actionParameters = actionExecutionContext.getParameters();
    addServiceCheckAction(stage, preferredHostName, smokeTestRole, nowTimestamp, serviceName, componentName,
        actionParameters, actionExecutionContext.isRetryAllowed(),
        actionExecutionContext.isFailureAutoSkipped(),false);
  }

  /**
   * Assuming all hosts are healthy and not in maintenance mode. Rank the hosts based on availability.
   * Let S = all hosts with 0 PENDING/RUNNING/QUEUED/IN-PROGRESS tasks
   * Let S' be all such other hosts.
   *
   * If S is non-empty, pick a random host from it. If S is empty and S' is non-empty, pick a random host from S'.
   * @param candidateHostNames All possible host names
   * @return Random host with a preference for those that are available to process commands immediately.
   */
  private String selectRandomHostNameWithPreferenceOnAvailability(List<String> candidateHostNames) throws AmbariException {
    if (null == candidateHostNames || candidateHostNames.isEmpty()) {
      return null;
    }
    if (candidateHostNames.size() == 1) {
      return candidateHostNames.get(0);
    }

    List<String> hostsWithZeroCommands = new ArrayList<>();
    List<String> hostsWithInProgressCommands = new ArrayList<>();

    Map<Long, Integer> hostIdToCount = hostRoleCommandDAO.getHostIdToCountOfCommandsWithStatus(HostRoleStatus.IN_PROGRESS_STATUSES);
    for (String hostName : candidateHostNames) {
      Host host = clusters.getHost(hostName);

      if (hostIdToCount.containsKey(host.getHostId()) && hostIdToCount.get(host.getHostId()) > 0) {
        hostsWithInProgressCommands.add(hostName);
      } else {
        hostsWithZeroCommands.add(hostName);
      }
    }

    List<String> preferredList = !hostsWithZeroCommands.isEmpty() ? hostsWithZeroCommands : hostsWithInProgressCommands;
    if (!preferredList.isEmpty()) {
      int randomIndex = new Random().nextInt(preferredList.size());
      return preferredList.get(randomIndex);
    }

    return null;
  }

  /**
   * Creates and populates service check EXECUTION_COMMAND for host. Not all
   * EXECUTION_COMMAND parameters are populated here because they are not needed
   * by service check.
   */
  public void addServiceCheckAction(Stage stage, String hostname, String smokeTestRole,
      long nowTimestamp, String serviceName, String componentName,
      Map<String, String> actionParameters, boolean retryAllowed, boolean autoSkipFailure, boolean useLatestConfigs)
          throws AmbariException {

    String clusterName = stage.getClusterName();
    Cluster cluster = clusters.getCluster(clusterName);
    Service service = cluster.getService(serviceName);
    ServiceComponent component = null;
    if (null != componentName) {
      component = service.getServiceComponent(componentName);
    }
    StackId stackId = (null != component) ? component.getDesiredStackId() : service.getDesiredStackId();

    AmbariMetaInfo ambariMetaInfo = managementController.getAmbariMetaInfo();
    ServiceInfo serviceInfo = ambariMetaInfo.getService(stackId.getStackName(),
        stackId.getStackVersion(), serviceName);

    stage.addHostRoleExecutionCommand(hostname, Role.valueOf(smokeTestRole),
        RoleCommand.SERVICE_CHECK,
        new ServiceComponentHostOpInProgressEvent(componentName, hostname, nowTimestamp),
        cluster.getClusterName(), serviceName, retryAllowed, autoSkipFailure);

    HostRoleCommand hrc = stage.getHostRoleCommand(hostname, smokeTestRole);
    if (hrc != null) {
      hrc.setCommandDetail(String.format("%s %s", RoleCommand.SERVICE_CHECK.toString(), serviceName));
    }
    // [ type -> [ key, value ] ]
    Map<String, Map<String, String>> configurations =
