class ConfigureAction extends AbstractUpgradeServerAction {

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

    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 = getClusters().getCluster(clusterName);
    UpgradeContext upgradeContext = getUpgradeContext(cluster);

    // such as hdfs-site or hbase-env
    String configType = commandParameters.get(ConfigureTask.PARAMETER_CONFIG_TYPE);
    String serviceName = cluster.getServiceByConfigType(configType);

    // !!! we couldn't get the service based on its config type, so try the associated
    if (StringUtils.isBlank(serviceName)) {
      serviceName = commandParameters.get(ConfigureTask.PARAMETER_ASSOCIATED_SERVICE);
    }

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

    // extract setters
    List<ConfigurationKeyValue> keyValuePairs = Collections.emptyList();
    String keyValuePairJson = commandParameters.get(ConfigureTask.PARAMETER_KEY_VALUE_PAIRS);
    if (null != keyValuePairJson) {
      keyValuePairs = m_gson.fromJson(
          keyValuePairJson, new TypeToken<List<ConfigurationKeyValue>>(){}.getType());
      keyValuePairs = getAllowedSets(cluster, configType, keyValuePairs);
    }

    // extract transfers
    List<Transfer> transfers = Collections.emptyList();
    String transferJson = commandParameters.get(ConfigureTask.PARAMETER_TRANSFERS);
    if (null != transferJson) {
      transfers = m_gson.fromJson(transferJson, new TypeToken<List<Transfer>>(){}.getType());
      transfers = getAllowedTransfers(cluster, configType, transfers);
    }

    // extract replacements
    List<Replace> replacements = Collections.emptyList();
    String replaceJson = commandParameters.get(ConfigureTask.PARAMETER_REPLACEMENTS);
    if (null != replaceJson) {
      replacements = m_gson.fromJson(replaceJson, new TypeToken<List<Replace>>(){}.getType());
      replacements = getAllowedReplacements(cluster, configType, replacements);
    }

    // extract insertions
    List<Insert> insertions = Collections.emptyList();
    String insertJson = commandParameters.get(ConfigureTask.PARAMETER_INSERTIONS);
    if (null != insertJson) {
      insertions = m_gson.fromJson(insertJson, new TypeToken<List<Insert>>(){}.getType());
      insertions = getAllowedInsertions(cluster, configType, insertions);
    }

    // if there is nothing to do, then skip the task
    if (keyValuePairs.isEmpty() && transfers.isEmpty() && replacements.isEmpty() && insertions.isEmpty()) {
      String message = "cluster={0}, type={1}, transfers={2}, replacements={3}, insertions={4}, configurations={5}";
      message = MessageFormat.format(message, clusterName, configType, transfers, replacements,
          insertions, keyValuePairs);

      StringBuilder buffer = new StringBuilder(
          "Skipping this configuration task since none of the conditions were met and there are no transfers, replacements, or insertions.").append("\n");

      buffer.append(message);

      return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", buffer.toString(), "");
    }

    // if only 1 of the required properties was null and no transfer properties,
    // then something went wrong
    if (null == clusterName || null == configType
        || (keyValuePairs.isEmpty() && transfers.isEmpty() && replacements.isEmpty() && insertions.isEmpty())) {
      String message = "cluster={0}, type={1}, transfers={2}, replacements={3}, insertions={4}, configurations={5}";

      message = MessageFormat.format(message, clusterName, configType, transfers, replacements,
          insertions, keyValuePairs);

      return createCommandReport(0, HostRoleStatus.FAILED, "{}", "", message);
    }

    Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs();
    DesiredConfig desiredConfig = desiredConfigs.get(configType);
    if (desiredConfig == null) {
      throw new AmbariException("Could not find desired config type with name " + configType);
    }

    Config config = cluster.getConfig(configType, desiredConfig.getTag());
    if (config == null) {
      throw new AmbariException("Could not find config type with name " + configType);
    }

    StackId configStack = config.getStackId();

    // !!! initial reference values
    Map<String, String> base = config.getProperties();
    Map<String, String> newValues = new HashMap<>(base);

    boolean changedValues = false;

    // !!! do transfers first before setting defined values
    StringBuilder outputBuffer = new StringBuilder(250);
    for (Transfer transfer : transfers) {
      switch (transfer.operation) {
        case COPY:
          String valueToCopy = null;
          if( null == transfer.fromType ) {
            // copying from current configuration
            valueToCopy = base.get(transfer.fromKey);
          } else {
            // copying from another configuration
            Config other = cluster.getDesiredConfigByType(transfer.fromType);
            if (null != other){
              Map<String, String> otherValues = other.getProperties();
              if (otherValues.containsKey(transfer.fromKey)){
                valueToCopy = otherValues.get(transfer.fromKey);
              }
            }
          }

          // if the value is null use the default if it exists
          if (StringUtils.isBlank(valueToCopy) && !StringUtils.isBlank(transfer.defaultValue)) {
            valueToCopy = transfer.defaultValue;
          }

          if (StringUtils.isNotBlank(valueToCopy)) {
            // possibly coerce the value on copy
            if (transfer.coerceTo != null) {
              switch (transfer.coerceTo) {
                case YAML_ARRAY: {
                  // turn c6401,c6402 into ['c6401',c6402']
                  String[] splitValues = StringUtils.split(valueToCopy, ',');
                  List<String> quotedValues = new ArrayList<>(splitValues.length);
                  for (String splitValue : splitValues) {
                    quotedValues.add("'" + StringUtils.trim(splitValue) + "'");
                  }

                  valueToCopy = "[" + StringUtils.join(quotedValues, ',') + "]";

                  break;
                }
                default:
                  break;
              }
            }

            // at this point we know that we have a changed value
            changedValues = true;
            newValues.put(transfer.toKey, valueToCopy);

            // append standard output
            updateBufferWithMessage(outputBuffer, MessageFormat.format("Created {0}/{1} = \"{2}\"",
                configType,
                transfer.toKey, mask(transfer, valueToCopy)));
          }
          break;
        case MOVE:
          // if the value existed previously, then update the maps with the new
          // key; otherwise if there is a default value specified, set the new
          // key with the default
          if (newValues.containsKey(transfer.fromKey)) {
            newValues.put(transfer.toKey, newValues.remove(transfer.fromKey));
            changedValues = true;

            // append standard output
            updateBufferWithMessage(outputBuffer,
                MessageFormat.format("Renamed {0}/{1} to {2}/{3}", configType,
                transfer.fromKey, configType, transfer.toKey));

          } else if (StringUtils.isNotBlank(transfer.defaultValue)) {
            newValues.put(transfer.toKey, transfer.defaultValue);
            changedValues = true;

            // append standard output
            updateBufferWithMessage(outputBuffer,
                MessageFormat.format("Created {0}/{1} with default value \"{2}\"",
                configType, transfer.toKey, mask(transfer, transfer.defaultValue)));
          }

          break;
        case DELETE:
          if (ALL_SYMBOL.equals(transfer.deleteKey)) {
            newValues.clear();

            // append standard output
            updateBufferWithMessage(outputBuffer,
                MessageFormat.format("Deleted all keys from {0}", configType));

            for (String keeper : transfer.keepKeys) {
              if (base.containsKey(keeper) && base.get(keeper) != null) {
                newValues.put(keeper, base.get(keeper));

                // append standard output
                updateBufferWithMessage(outputBuffer,
                    MessageFormat.format("Preserved {0}/{1} after delete", configType, keeper));
              }
            }

            // !!! with preserved edits, find the values that are different from
            // the stack-defined and keep them - also keep values that exist in
            // the config but not on the stack
            if (transfer.preserveEdits) {
              List<String> edited = findValuesToPreserve(clusterName, config);
              for (String changed : edited) {
                newValues.put(changed, base.get(changed));

                // append standard output
                updateBufferWithMessage(outputBuffer,
                    MessageFormat.format("Preserved {0}/{1} after delete",
                    configType, changed));
              }
            }

            changedValues = true;
          } else {
            newValues.remove(transfer.deleteKey);
            changedValues = true;

            // append standard output
            updateBufferWithMessage(outputBuffer,
                MessageFormat.format("Deleted {0}/{1}", configType,
                transfer.deleteKey));
          }

          break;
      }
    }

    // set all key/value pairs
    if (null != keyValuePairs && !keyValuePairs.isEmpty()) {
      for (ConfigurationKeyValue keyValuePair : keyValuePairs) {
        String key = keyValuePair.key;
        String value = keyValuePair.value;

        if (null != key) {
          String oldValue = base.get(key);

          // !!! values are not changing, so make this a no-op
          if (StringUtils.equals(value, oldValue)) {
            if (sourceStackId.equals(targetStackId) && !changedValues) {
              updateBufferWithMessage(outputBuffer,
                  MessageFormat.format(
                  "{0}/{1} for cluster {2} would not change, skipping setting", configType, key,
                  clusterName));

              // continue because this property is not changing
              continue;
            }
          }

          // !!! only put a key/value into this map of new configurations if
          // there was a key, otherwise this will put something like null=null
          // into the configs which will cause NPEs after upgrade - this is a
          // byproduct of the configure being able to take a list of transfers
          // without a key/value to set
          newValues.put(key, value);

          final String message;
          if (StringUtils.isEmpty(value)) {
            message = MessageFormat.format("{0}/{1} changed to an empty value", configType, key);
          } else {
            message = MessageFormat.format("{0}/{1} changed to \"{2}\"", configType, key,
                mask(keyValuePair, value));
          }

          updateBufferWithMessage(outputBuffer, message);
        }
      }
    }

    // replacements happen only on the new values (as they are initialized from
    // the existing pre-upgrade values)
    for (Replace replacement : replacements) {
      // the key might exist but might be null, so we need to check this
      // condition when replacing a part of the value
      String toReplace = newValues.get(replacement.key);
      if (StringUtils.isNotBlank(toReplace)) {
        if (!toReplace.contains(replacement.find)) {
          updateBufferWithMessage(outputBuffer,
              MessageFormat.format("String \"{0}\" was not found in {1}/{2}",
              replacement.find, configType, replacement.key));
        } else {
          String replaced = StringUtils.replace(toReplace, replacement.find, replacement.replaceWith);

          newValues.put(replacement.key, replaced);

          // customize the replacement message if the new value is empty
          if (StringUtils.isEmpty(replacement.replaceWith)) {
            updateBufferWithMessage(outputBuffer, MessageFormat.format(
                "Removed \"{0}\" from {1}/{2}", replacement.find, configType, replacement.key));
          } else {
            updateBufferWithMessage(outputBuffer,
                MessageFormat.format("Replaced {0}/{1} containing \"{2}\" with \"{3}\"", configType,
                    replacement.key, replacement.find, replacement.replaceWith));
          }
        }
      } else {
        updateBufferWithMessage(outputBuffer, MessageFormat.format(
            "Skipping replacement for {0}/{1} because it does not exist or is empty.",
            configType, replacement.key));
      }
    }

    // insertions happen only on the new values (as they are initialized from
    // the existing pre-upgrade values)
    for (Insert insert : insertions) {
      String valueToInsertInto = newValues.get(insert.key);

      // if the key doesn't exist, then do no work
      if (StringUtils.isNotBlank(valueToInsertInto)) {
        // make this insertion idempotent - don't do it if the value already
        // contains the content
        if (StringUtils.contains(valueToInsertInto, insert.value)) {
          updateBufferWithMessage(outputBuffer,
              MessageFormat.format("Skipping insertion for {0}/{1} because it already contains {2}",
                  configType, insert.key, insert.value));

          continue;
        }

        // new line work
        String valueToInsert = insert.value;
        if (insert.newlineBefore) {
          valueToInsert = System.lineSeparator() + valueToInsert;
        }

        // new line work
        if (insert.newlineAfter) {
          valueToInsert = valueToInsert + System.lineSeparator();
        }

        switch (insert.insertType) {
          case APPEND:
            valueToInsertInto = valueToInsertInto + valueToInsert;
            break;
          case PREPEND:
            valueToInsertInto = valueToInsert + valueToInsertInto;
            break;
          default:
            LOG.error("Unable to insert {0}/{1} with unknown insertion type of {2}", configType,
                insert.key, insert.insertType);
            break;
        }

        newValues.put(insert.key, valueToInsertInto);

        updateBufferWithMessage(outputBuffer, MessageFormat.format(
            "Updated {0}/{1} by inserting \"{2}\"", configType, insert.key, insert.value));
      } else {
        updateBufferWithMessage(outputBuffer, MessageFormat.format(
            "Skipping insertion for {0}/{1} because it does not exist or is empty.", configType,
            insert.key));
      }
    }

    // !!! check to see if we're going to a new stack and double check the
    // configs are for the target.  Then simply update the new properties instead
    // of creating a whole new history record since it was already done
    if (!targetStackId.equals(sourceStackId) && targetStackId.equals(configStack)) {
      config.setProperties(newValues);
      config.save();

      m_metadataHolder.get().updateData(m_ambariManagementController.get().getClusterMetadataOnConfigsUpdate(cluster));
      m_agentConfigsHolder.get().updateData(cluster.getClusterId(), null);

      return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", outputBuffer.toString(), "");
    }

    // !!! values are different and within the same stack.  create a new
    // config and service config version
    Direction direction = upgradeContext.getDirection();
    String serviceVersionNote = String.format("%s %s %s", direction.getText(true),
        direction.getPreposition(), upgradeContext.getRepositoryVersion().getVersion());

    String auditName = getExecutionCommand().getRoleParams().get(ServerAction.ACTION_USER_NAME);

    if (auditName == null) {
      auditName = m_configuration.getAnonymousAuditName();
    }

    m_configHelper.createConfigType(cluster, targetStackId, m_controller, configType,
        newValues, auditName, serviceVersionNote);

    return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", outputBuffer.toString(), "");
  }

}