class CreateAndConfigureAction extends ConfigureAction {

  @Override
  public CommandReport execute(
      ConcurrentMap<String, Object> requestSharedDataContext)
      throws AmbariException, InterruptedException {

    LOG.info("Create and Configure...");

    Map<String,String> commandParameters = getCommandParameters();
    if( null == commandParameters || commandParameters.isEmpty() ){
      return createCommandReport(0, HostRoleStatus.FAILED, "{}", "",
          "Unable to change configuration values without command parameters");
    }

    String clusterName = commandParameters.get("clusterName");
    Cluster cluster = m_clusters.getCluster(clusterName);
    UpgradeContext upgradeContext = getUpgradeContext(cluster);

    Direction direction = upgradeContext.getDirection();
    if (direction == Direction.DOWNGRADE) {
      return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", "", "Skip changing configuration values for downgrade");
    }

    String configType = commandParameters.get(CreateAndConfigureTask.PARAMETER_CONFIG_TYPE);
    String serviceName = cluster.getServiceByConfigType(configType);

    if (StringUtils.isBlank(serviceName)) {
      serviceName = commandParameters.get(CreateAndConfigureTask.PARAMETER_ASSOCIATED_SERVICE);
    }

    RepositoryVersionEntity sourceRepoVersion = upgradeContext.getSourceRepositoryVersion(serviceName);
    RepositoryVersionEntity targetRepoVersion = upgradeContext.getTargetRepositoryVersion(serviceName);
    StackId sourceStackId = sourceRepoVersion.getStackId();
    StackId targetStackId = targetRepoVersion.getStackId();

    if (!sourceStackId.equals(targetStackId)){
      return createCommandReport(0, HostRoleStatus.FAILED, "{}", "",
          "Unable to change configuration values across stacks. Use regular config task type instead.");
    }

    Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs();
    DesiredConfig desiredConfig = desiredConfigs.get(configType);
    if (desiredConfig == null) {
      LOG.info(String.format("Could not find desired config type with name %s. Create it with default values.", configType));

      // populate a map with default configurations from the new stack
      Map<String, Map<String, String>> newServiceDefaultConfigsByType = m_configHelper.getDefaultProperties(
          targetStackId, serviceName);

      if (!newServiceDefaultConfigsByType.containsKey(configType)){
        String error = String.format("%s in %s does not contain configuration type %s", serviceName, targetStackId.getStackId(), configType);
        LOG.error(error);
        return createCommandReport(0, HostRoleStatus.FAILED, "{}", "", error);
      }

      Map<String, String> defaultConfigsForType = newServiceDefaultConfigsByType.get(configType);
      // Remove any property for the new config type whose value is NULL
      Iterator<Map.Entry<String, String>> iter = defaultConfigsForType.entrySet().iterator();
      while (iter.hasNext()) {
        Map.Entry<String, String> entry = iter.next();
        if (entry.getValue() == null) {
          iter.remove();
        }
      }

      String serviceVersionNote = String.format("%s %s %s", direction.getText(true),
          direction.getPreposition(), upgradeContext.getRepositoryVersion().getVersion());

      m_configHelper.createConfigType(cluster, targetStackId,
          m_controller,
          configType, defaultConfigsForType,
          m_controller.getAuthName(), serviceVersionNote);
    }

    return super.execute(requestSharedDataContext);
  }

}