@Override
  public synchronized ConfigurationResponse createConfiguration(
      ConfigurationRequest request, boolean refreshCluster) throws AmbariException, AuthorizationException {
    if (null == request.getClusterName() || request.getClusterName().isEmpty()
        || null == request.getType() || request.getType().isEmpty()
        || null == request.getProperties()) {
      throw new IllegalArgumentException("Invalid Arguments,"
          + " clustername, config type and configs should not"
          + " be null or empty");
    }

    Cluster cluster = clusters.getCluster(request.getClusterName());

    String configType = request.getType();

    // If the config type is for a service, then allow a user with SERVICE_MODIFY_CONFIGS to
    // update, else ensure the user has CLUSTER_MODIFY_CONFIGS
    String service = cluster.getServiceByConfigType(configType);


    // Get the changes so that the user's intention can be determined. For example, maybe
    // the user wants to change the run-as user for a service or maybe the the cluster-wide
    // recovery mode setting.
    Map<String, String[]> propertyChanges = getPropertyChanges(cluster, request);

    if(StringUtils.isEmpty(service)) {
      // If the configuration is not attached to a specific service, it is a cluster-wide configuration
      // type. For example, cluster-env.

      // If the user is trying to set the cluster-wide recovery mode, ensure that user
      // has the appropriate authorization
      validateAuthorizationToManageServiceAutoStartConfiguration(cluster, configType, propertyChanges);

      // If the user is trying to set any other cluster-wide property, ensure that user
      // has the appropriate authorization
      validateAuthorizationToModifyConfigurations(cluster, configType, propertyChanges,
          Collections.singletonMap("cluster-env", Collections.singleton("recovery_enabled")),
          false);
    }
    else {
      // If the user is trying to set any service-level property, ensure that user
      // has the appropriate authorization
      validateAuthorizationToModifyConfigurations(cluster, configType, propertyChanges, null, true);

      // Ensure the user is allowed to update service users and groups.
      validateAuthorizationToUpdateServiceUsersAndGroups(cluster, configType, propertyChanges);
    }

    Map<String, String> requestProperties = request.getProperties();

    // Configuration attributes are optional. If not present, use default(provided by stack), otherwise merge default
    // with request-provided
    Map<String, Map<String, String>> requestPropertiesAttributes = request.getPropertiesAttributes();

    if (requestPropertiesAttributes != null && requestPropertiesAttributes.containsKey(PASSWORD)) {
      for (Map.Entry<String, String> requestEntry : requestPropertiesAttributes.get(PASSWORD).entrySet()) {
        String passwordProperty = requestEntry.getKey();
        if(requestProperties.containsKey(passwordProperty) && requestEntry.getValue().equals("true")) {
          String passwordPropertyValue = requestProperties.get(passwordProperty);
          if (!SecretReference.isSecret(passwordPropertyValue)) {
            continue;
          }
          SecretReference ref = new SecretReference(passwordPropertyValue, cluster);
          String refValue = ref.getValue();
          requestProperties.put(passwordProperty, refValue);
        }
      }
    }

    Map<PropertyInfo.PropertyType, Set<String>> propertiesTypes = cluster.getConfigPropertiesTypes(request.getType());
    if(propertiesTypes.containsKey(PropertyType.PASSWORD)) {
      for(String passwordProperty : propertiesTypes.get(PropertyType.PASSWORD)) {
        if(requestProperties.containsKey(passwordProperty)) {
          String passwordPropertyValue = requestProperties.get(passwordProperty);
          if (!SecretReference.isSecret(passwordPropertyValue)) {
            continue;
          }
          SecretReference ref = new SecretReference(passwordPropertyValue, cluster);
          String refValue = ref.getValue();
          requestProperties.put(passwordProperty, refValue);
        }
      }
    }



    Map<String, Config> configs = cluster.getConfigsByType(
        request.getType());
    if (null == configs) {
      configs = new HashMap<>();
    }

    Map<String, Map<String, String>> propertiesAttributes = new HashMap<>();

    Set<StackId> visitedStacks = new HashSet<>();

    for (Service clusterService : cluster.getServices().values()) {
      StackId stackId = clusterService.getDesiredStackId();
      StackInfo stackInfo = ambariMetaInfo.getStack(clusterService.getDesiredStackId());

      if (visitedStacks.contains(stackId)) {
        continue;
      }

      Map<String, Map<String, String>> defaultConfigAttributes = stackInfo.getDefaultConfigAttributesForConfigType(configType);
      if (null != defaultConfigAttributes) {
        ConfigHelper.mergeConfigAttributes(propertiesAttributes, defaultConfigAttributes);
      }

      visitedStacks.add(stackId);
    }

    // overwrite default attributes with request attributes
    if(requestPropertiesAttributes != null){
      ConfigHelper.mergeConfigAttributes(propertiesAttributes, requestPropertiesAttributes);
    }

    if (configs.containsKey(request.getVersionTag())) {
      throw new AmbariException(MessageFormat.format("Configuration with tag ''{0}'' exists for ''{1}''",
          request.getVersionTag(),
          request.getType()));
    }

    StackId stackId = null;
    if (null != service) {
      try {
        Service svc = cluster.getService(service);
        stackId = svc.getDesiredStackId();
      } catch (AmbariException ambariException) {
        LOG.warn("Adding configurations for {} even though its parent service {} is not installed",
            configType, service);
      }
    }

    if (null == stackId) {
      stackId = cluster.getDesiredStackVersion();
    }

    Config config = createConfig(cluster, stackId, request.getType(), requestProperties,
      request.getVersionTag(), propertiesAttributes, refreshCluster);

    LOG.info(MessageFormat.format("Creating configuration with tag ''{0}'' to cluster ''{1}''  for configuration type {2}",
        request.getVersionTag(),
        request.getClusterName(),
        configType));

    return new ConfigurationResponse(cluster.getClusterName(), config);
  }
Creates a configuration.
<p>
This implementation ensures the authenticated user is authorized to create the new configuration
based on the details of what properties are being changed and the authorizations the authenticated
user has been granted.
<p>
Example
<ul>
<li>
If the user is attempting to change a service-level configuration that user must be granted the
<code>SERVICE_MODIFY_CONFIGS</code> privilege (authorization)
</li>
<li>
If the user is attempting to change the cluster-wide value to enable or disable auto-start
(<code>cluster-env/recovery_enabled</code>), that user must be granted the
<code>CLUSTER_MANAGE_AUTO_START</code> privilege (authorization)
</li>
</ul>

@param request the request object which defines the configuration.
@throws AmbariException when the configuration cannot be created.
ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java