AMBARI-8976. Use cluster property rather than cluster-env/security_enabled to enable or disable Kerberos. (rlevas via yusaku)

ambari-server/docs/api/v1/clusters-cluster.md

@@ -83,6 +83,7 @@ Get information for the cluster "cluster001".
 			"Host/host_status/ALERT" : 0
 		},
 		"provisioning_state" : "INIT",
+		"security_type" : "NONE",
 		"total_hosts" : 3,
 		"version" : "HDP-2.0",
 		"desired_configs" : {

ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java

@@ -103,12 +103,14 @@ import org.apache.ambari.server.state.Config;
 import org.apache.ambari.server.state.ConfigFactory;
 import org.apache.ambari.server.state.ConfigHelper;
 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.OperatingSystemInfo;
 import org.apache.ambari.server.state.PropertyInfo;
 import org.apache.ambari.server.state.PropertyInfo.PropertyType;
 import org.apache.ambari.server.state.RepositoryInfo;
+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.ServiceComponentFactory;
@@ -145,7 +147,6 @@ import com.google.inject.Singleton;
 import com.google.inject.persist.Transactional;
 
 import org.apache.ambari.server.controller.internal.RequestResourceFilter;
-import org.apache.ambari.server.state.HostComponentAdminState;
 
 @Singleton
 public class AmbariManagementControllerImpl implements AmbariManagementController {
@@ -1160,6 +1161,8 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
   private synchronized RequestStatusResponse updateCluster(ClusterRequest request)
       throws AmbariException {
 
+    RequestStageContainer requestStageContainer = null;
+
     if (request.getClusterId() == null
         && (request.getClusterName() == null
         || request.getClusterName().isEmpty())) {
@@ -1169,6 +1172,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     LOG.info("Received a updateCluster request"
         + ", clusterId=" + request.getClusterId()
         + ", clusterName=" + request.getClusterName()
+        + ", securityType=" + request.getSecurityType()
         + ", request=" + request);
 
     final Cluster cluster;
@@ -1196,40 +1200,6 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
       cluster.setClusterName(request.getClusterName());
     }
 
-    // ----------------------
-    // Check to see if the security state is being changed... if so, attempt to enable or disable
-    // Kerberos
-    boolean toggleKerberos = false;
-
-    String desiredSecurityState = null;
-    List<ConfigurationRequest> desiredConfig = request.getDesiredConfig();
-    if (desiredConfig != null) {
-      for (ConfigurationRequest configurationRequest : desiredConfig) {
-        if ("cluster-env".equals(configurationRequest.getType())) {
-          Map<String, String> properties = configurationRequest.getProperties();
-
-          if ((properties == null) || properties.isEmpty()) {
-            Config configClusterEnv = cluster.getConfig(configurationRequest.getType(), configurationRequest.getVersionTag());
-            if (configClusterEnv != null) {
-              properties = configClusterEnv.getProperties();
-            }
-          }
-
-          desiredSecurityState = (properties == null) ? null : properties.get("security_enabled");
-        }
-      }
-    }
-
-    if(desiredSecurityState != null) {
-      Config configClusterEnv = cluster.getDesiredConfigByType("cluster-env");
-      Map<String, String> clusterEnvProperties = (configClusterEnv == null) ? null : configClusterEnv.getProperties();
-      if (clusterEnvProperties != null) {
-        toggleKerberos = !desiredSecurityState.equals(clusterEnvProperties.get("security_enabled"));
-      }
-    }
-    // ----------------------
-
-
     // set or create configuration mapping (and optionally create the map of properties)
     if (null != request.getDesiredConfig()) {
       Set<Config> configs = new HashSet<Config>();
@@ -1346,7 +1316,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
       }
 
       ClusterResponse clusterResponse =
-          new ClusterResponse(cluster.getClusterId(), cluster.getClusterName(), null, null, null, null, null);
+          new ClusterResponse(cluster.getClusterId(), cluster.getClusterName(), null, null, null, null, null, null);
 
       Map<String, Collection<ServiceConfigVersionResponse>> map =
         new HashMap<String, Collection<ServiceConfigVersionResponse>>();
@@ -1359,16 +1329,22 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
       saveClusterUpdate(request, clusterResponse);
     }
 
-    RequestStageContainer requestStageContainer = null;
-    if(toggleKerberos) {
-      Map<String, Service> services = cluster.getServices();
-      if ((services != null) && services.containsKey("KERBEROS")) {
-        // Handle either adding or removing Kerberos from the cluster. This may generate multiple stages
-        // or not depending the current state of the cluster.  The main configuration used to determine
-        // whether Kerberos is to be added or removed is cluster-config/security_enabled.
-        requestStageContainer = kerberosHelper.toggleKerberos(cluster,
-            request.getKerberosDescriptor(), null);
-      }
+    // set the new security type of the cluster if change is requested
+    SecurityType securityType = request.getSecurityType();
+    if((securityType != null) && (cluster.getSecurityType() != securityType) ){
+        LOG.info("Received cluster security type change request from {} to {}",
+            cluster.getSecurityType().name(), securityType.name());
+
+        if ((securityType == SecurityType.KERBEROS) || (securityType == SecurityType.NONE)) {
+          // Since the security state of the cluster has changed, invoke toggleKerberos to handle
+          // adding or removing Kerberos from the cluster. This may generate multiple stages
+          // or not depending the current state of the cluster.
+          requestStageContainer = kerberosHelper.toggleKerberos(cluster, securityType, request.getKerberosDescriptor(), requestStageContainer);
+        } else {
+          throw new IllegalArgumentException(String.format("Unexpected security type encountered: %s", securityType.name()));
+        }
+
+        cluster.setSecurityType(securityType);
     }
 
     if (requestStageContainer != null) {

ambari-server/src/main/java/org/apache/ambari/server/controller/ClusterRequest.java

@@ -18,6 +18,7 @@
 
 package org.apache.ambari.server.controller;
 
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.kerberos.KerberosDescriptor;
 
 import java.util.List;
@@ -36,9 +37,16 @@ public class ClusterRequest {
   private String stackVersion; // for CREATE/UPDATE
 
   private String provisioningState; // for GET/CREATE/UPDATE
-  
+
+  /**
+   * The cluster's security type
+   * <p/>
+   * See {@link org.apache.ambari.server.state.SecurityType} for relevant values.
+   */
+  private SecurityType securityType; // for GET/CREATE/UPDATE
+
   Set<String> hostNames; // CREATE/UPDATE
-  
+
   private List<ConfigurationRequest> configs = null;
 
   private ServiceConfigVersionRequest serviceConfigVersionRequest = null;
@@ -58,22 +66,23 @@ public class ClusterRequest {
 
   public ClusterRequest(Long clusterId, String clusterName,
       String stackVersion, Set<String> hostNames) {
-    this(clusterId, clusterName, null, stackVersion, hostNames);
-  }  
-  
-  public ClusterRequest(Long clusterId, String clusterName, 
-      String provisioningState, String stackVersion, Set<String> hostNames) {
-    this(clusterId, clusterName, provisioningState, stackVersion, hostNames, null, null);
+    this(clusterId, clusterName, null, null, stackVersion, hostNames);
+  }
+
+  public ClusterRequest(Long clusterId, String clusterName,
+      String provisioningState, SecurityType securityType, String stackVersion, Set<String> hostNames) {
+    this(clusterId, clusterName, provisioningState, securityType, stackVersion, hostNames, null, null);
   }
 
   public ClusterRequest(Long clusterId, String clusterName,
-                        String provisioningState, String stackVersion,
+                        String provisioningState, SecurityType securityType, String stackVersion,
                         Set<String> hostNames, KerberosDescriptor kerberosDescriptor,
                         Map<String, Object> sessionAttributes) {
     super();
     this.clusterId         = clusterId;
     this.clusterName       = clusterName;
     this.provisioningState = provisioningState;
+    this.securityType      = securityType;
     this.stackVersion      = stackVersion;
     this.hostNames         = hostNames;
     this.sessionAttributes = sessionAttributes;
@@ -96,22 +105,22 @@ public class ClusterRequest {
   public String getClusterName() {
     return clusterName;
   }
-  
+
   /**
    * Gets whether the cluster is still initializing or has finished with its
    * deployment requests.
-   * 
+   *
    * @return either {@code INIT} or {@code INSTALLED} or {@code null} if not set
    *         on the request.
    */
   public String getProvisioningState(){
     return provisioningState;
   }
-  
+
   /**
    * Sets whether the cluster is still initializing or has finished with its
    * deployment requests.
-   * 
+   *
    * @param provisioningState
    *          either {@code INIT} or {@code INSTALLED}, or {@code null} if not
    *          set on the request.
@@ -120,6 +129,29 @@ public class ClusterRequest {
     this.provisioningState = provisioningState;
   }
 
+  /**
+   * Gets the cluster's security type.
+   * <p/>
+   * See {@link org.apache.ambari.server.state.SecurityType} for relevant values.
+   *
+   * @return a SecurityType declaring the security type; or {@code null} if not set set on the request
+   */
+  public SecurityType getSecurityType() {
+    return securityType;
+  }
+
+  /**
+   * Sets the cluster's security type.
+   * <p/>
+   * See {@link org.apache.ambari.server.state.SecurityType} for relevant values.
+   *
+   * @param securityType a SecurityType declaring the cluster's security type; or {@code null} if not
+   *                     set on the request
+   */
+  public void setSecurityType(SecurityType securityType) {
+    this.securityType = securityType;
+  }
+
   /**
    * @return the stackVersion
    */
@@ -155,7 +187,7 @@ public class ClusterRequest {
   public void setHostNames(Set<String> hostNames) {
     this.hostNames = hostNames;
   }
-  
+
   /**
    * Sets the configs requests (if any).
    *
@@ -164,7 +196,7 @@ public class ClusterRequest {
   public void setDesiredConfig(List<ConfigurationRequest> configRequests) {
     configs = configRequests;
   }
-  
+
   /**
    * Gets any configuration-based request (if any).
    * @return the list of configuration requests,
@@ -198,6 +230,7 @@ public class ClusterRequest {
     sb.append("{" + " clusterName=").append(clusterName)
         .append(", clusterId=").append(clusterId)
         .append(", provisioningState=").append(provisioningState)
+        .append(", securityType=").append(securityType)
         .append(", stackVersion=").append(stackVersion)
         .append(", desired_scv=").append(serviceConfigVersionRequest)
         .append(", hosts=[");

ambari-server/src/main/java/org/apache/ambari/server/controller/ClusterResponse.java

@@ -25,6 +25,7 @@ import java.util.Set;
 
 import com.google.common.collect.Multimap;
 import org.apache.ambari.server.state.DesiredConfig;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.State;
 import org.apache.ambari.server.state.ClusterHealthReport;
 
@@ -41,16 +42,23 @@ public class ClusterResponse {
   private Map<String, DesiredConfig> desiredConfigs;
 
   private Map<String, Collection<ServiceConfigVersionResponse>> desiredServiceConfigVersions;
-  
+
   private String provisioningState;
 
+  /**
+   * The cluster's security.
+   * <p/>
+   * See {@link org.apache.ambari.server.state.SecurityType} for relevant values.
+   */
+  private String securityType;
+
   private Integer totalHosts;
 
   private ClusterHealthReport clusterHealthReport;
-  
+
   public ClusterResponse(Long clusterId, String clusterName,
-    State provisioningState, Set<String> hostNames, Integer totalHosts,
-    String desiredStackVersion, ClusterHealthReport clusterHealthReport) {
+                         State provisioningState, SecurityType securityType, Set<String> hostNames, Integer totalHosts,
+                         String desiredStackVersion, ClusterHealthReport clusterHealthReport) {
 
     super();
     this.clusterId = clusterId;
@@ -59,9 +67,16 @@ public class ClusterResponse {
     this.totalHosts = totalHosts;
     this.desiredStackVersion = desiredStackVersion;
     this.clusterHealthReport = clusterHealthReport;
-    
-    if (null != provisioningState)
+
+    if (null != provisioningState) {
       this.provisioningState = provisioningState.name();
+    }
+
+    if (null == securityType) {
+      this.securityType = SecurityType.NONE.name();
+    } else {
+      this.securityType = securityType.name();
+    }
   }
 
   /**
@@ -84,17 +99,39 @@ public class ClusterResponse {
   public Set<String> getHostNames() {
     return hostNames;
   }
-  
+
   /**
    * Gets whether the cluster is still initializing or has finished with its
    * deployment requests.
-   * 
+   *
    * @return either {@code INIT} or {@code INSTALLED}, never {@code null}.
    */
-  public String getProvisioningState(){
+  public String getProvisioningState() {
     return provisioningState;
   }
 
+  /**
+   * Gets the cluster's security type.
+   * <p/>
+   * See {@link org.apache.ambari.server.state.SecurityType} for relevant values.
+   *
+   * @return the cluster's security type
+   */
+  public String getSecurityType() {
+    return securityType;
+  }
+
+  /**
+   * Sets the cluster's security type.
+   * <p/>
+   * See {@link org.apache.ambari.server.state.SecurityType} for relevant values.
+   *
+   * @param securityType a String declaring the cluster's security type
+   */
+  public void setSecurityType(String securityType) {
+    this.securityType = securityType;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder();
@@ -105,7 +142,7 @@ public class ClusterResponse {
         + ", desiredStackVersion=" + desiredStackVersion
         + ", totalHosts=" + totalHosts
         + ", hosts=[");
-    
+
     if (hostNames != null) {
       int i = 0;
       for (String hostName : hostNames) {
@@ -124,8 +161,12 @@ public class ClusterResponse {
 
   @Override
   public boolean equals(Object o) {
-    if (this == o) return true;
-    if (o == null || getClass() != o.getClass()) return false;
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
 
     ClusterResponse that = (ClusterResponse) o;
 

ambari-server/src/main/java/org/apache/ambari/server/controller/KerberosHelper.java

@@ -43,6 +43,7 @@ import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
 import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.SecurityState;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.Service;
 import org.apache.ambari.server.state.ServiceComponentHost;
 import org.apache.ambari.server.state.StackId;
@@ -123,39 +124,46 @@ public class KerberosHelper {
   /**
    * Toggles Kerberos security to enable it or remove it depending on the state of the cluster.
    * <p/>
-   * The "cluster-env" configuration set is used to determine the security state of the cluster.
-   * If the "security_enabled" property is set to "true" than an attempt will be make to enable
-   * Kerberos; if "false" an attempt will be made to disable Kerberos.
-   * Also, the "kerberos_domain" is used as the default Kerberos realm for the cluster.
+   * The cluster "security_type" property is used to determine the security state of the cluster.
+   * If the declared security type is KERBEROS, than an attempt will be make to enable Kerberos; if
+   * the security type is NONE, an attempt will be made to disable Kerberos; otherwise, no operations
+   * will be performed.
    * <p/>
-   * The "krb5-conf" configuration type ios used to obtain information about the relevant KDC.
-   * The "kdc_type" property is used to determine what type of KDC is being used so that the
-   * appropriate actions maybe taken in order interact with it properly.
+   * It is expected that the "krb5-conf" configuration type is available.  It is used to obtain
+   * information about the relevant KDC.  For example, the "kdc_type" property is used to determine
+   * what type of KDC is being used so that the appropriate actions maybe taken in order interact
+   * with it properly.
+   * <p/>
+   * It is expected tht the "kerberos-env" configuration type is available.   It is used to obtain
+   * information about the Kerberos configuration, generally specific to the KDC being used.
    * <p/>
    * This process is idempotent such that it may be called several times, each time only affecting
    * items that need to be brought up to date.
    *
    * @param cluster               the relevant Cluster
+   * @param securityType          the SecurityType to handle; this value is expected to be either
+   *                              SecurityType.KERBEROS or SecurityType.NONE
    * @param kerberosDescriptor    a KerberosDescriptor containing updates to the descriptor already
    *                              configured for the cluster
    * @param requestStageContainer a RequestStageContainer to place generated stages, if needed -
-   *                              if null a new RequestStageContainer will be created.
-   * @return the updated or a new RequestStageContainer containing the stages that need to be
-   * executed to complete this task; or null if no stages need to be executed.
+   *                              if null a new RequestStageContainer will be created.   @return the updated or a new RequestStageContainer containing the stages that need to be
+   *                              executed to complete this task; or null if no stages need to be executed.
    * @throws AmbariException
    */
-  public RequestStageContainer toggleKerberos(Cluster cluster, KerberosDescriptor kerberosDescriptor,
+  public RequestStageContainer toggleKerberos(Cluster cluster, SecurityType securityType, KerberosDescriptor kerberosDescriptor,
                                               RequestStageContainer requestStageContainer)
       throws AmbariException {
 
     KerberosDetails kerberosDetails = getKerberosDetails(cluster);
 
-    if (kerberosDetails.isSecurityEnabled()) {
+    if (securityType == SecurityType.KERBEROS) {
       LOG.info("Configuring Kerberos for realm {} on cluster, {}", kerberosDetails.getDefaultRealm(), cluster.getClusterName());
       requestStageContainer = handle(cluster, kerberosDescriptor, kerberosDetails, null, null, requestStageContainer, enableKerberosHandler);
-    } else {
+    } else if (securityType == SecurityType.NONE) {
       LOG.info("Disabling Kerberos from cluster, {}", cluster.getClusterName());
       requestStageContainer = handle(cluster, kerberosDescriptor, kerberosDetails, null, null, requestStageContainer, disableKerberosHandler);
+    } else {
+      throw new AmbariException(String.format("Unexpected security type value: %s", securityType.name()));
     }
 
     return requestStageContainer;
@@ -166,6 +174,14 @@ public class KerberosHelper {
    * <p/>
    * No configurations will be altered as a result of this operation, however principals and keytabs
    * may be updated or created.
+   * <p/>
+   * It is expected that the "krb5-conf" configuration type is available.  It is used to obtain
+   * information about the relevant KDC.  For example, the "kdc_type" property is used to determine
+   * what type of KDC is being used so that the appropriate actions maybe taken in order interact
+   * with it properly.
+   * <p/>
+   * It is expected tht the "kerberos-env" configuration type is available.   It is used to obtain
+   * information about the Kerberos configuration, generally specific to the KDC being used.
    *
    * @param cluster                the relevant Cluster
    * @param kerberosDescriptor     a KerberosDescriptor containing updates to the descriptor already
@@ -466,6 +482,15 @@ public class KerberosHelper {
           }
         }
 
+        // Ensure the cluster-env/security_enabled flag is set properly
+        Map<String, String> clusterEnvProperties = kerberosConfigurations.get("cluster-env");
+        if (clusterEnvProperties == null) {
+          clusterEnvProperties = new HashMap<String, String>();
+          kerberosConfigurations.put("cluster-env", clusterEnvProperties);
+        }
+        clusterEnvProperties.put("security_enabled",
+            (kerberosDetails.getSecurityType() == SecurityType.KERBEROS) ? "true" : "false");
+
         // Always set up the necessary stages to perform the tasks needed to complete the operation.
         // Some stages may be no-ops, this is expected.
         // Gather data needed to create stages and tasks...
@@ -560,37 +585,6 @@ public class KerberosHelper {
       throw new AmbariException(message);
     }
 
-    Config configClusterEnv = cluster.getDesiredConfigByType("cluster-env");
-    if (configClusterEnv == null) {
-      String message = "The 'cluster-env' configuration is not available";
-      LOG.error(message);
-      throw new AmbariException(message);
-    }
-
-    Map<String, String> clusterEnvProperties = configClusterEnv.getProperties();
-    if (clusterEnvProperties == null) {
-      String message = "The 'cluster-env' configuration properties are not available";
-      LOG.error(message);
-      throw new AmbariException(message);
-    }
-
-    String securityEnabled = clusterEnvProperties.get("security_enabled");
-    if ((securityEnabled == null) || securityEnabled.isEmpty()) {
-      String message = "Missing 'securityEnabled' property of cluster-env, unable to determine the cluster's security state";
-      LOG.error(message);
-      throw new AmbariException(message);
-    }
-
-    if ("true".equalsIgnoreCase(securityEnabled)) {
-      kerberosDetails.setSecurityEnabled(true);
-    } else if ("false".equalsIgnoreCase(securityEnabled)) {
-      kerberosDetails.setSecurityEnabled(false);
-    } else {
-      String message = String.format("Invalid value for `security_enabled` property of cluster-env: %s", securityEnabled);
-      LOG.error(message);
-      throw new AmbariException(message);
-    }
-
     Config configKrb5Conf = cluster.getDesiredConfigByType("krb5-conf");
     if (configKrb5Conf == null) {
       String message = "The 'krb5-conf' configuration is not available";
@@ -635,6 +629,7 @@ public class KerberosHelper {
       }
     }
 
+    kerberosDetails.setSecurityType(cluster.getSecurityType());
     kerberosDetails.setDefaultRealm(krb5ConfProperties.get("realm"));
 
     // Set the KDCType to the the MIT_KDC as a fallback.
@@ -1510,19 +1505,10 @@ public class KerberosHelper {
    * configurations so they may be passed around more easily.
    */
   private static class KerberosDetails {
-    private boolean securityEnabled;
     private String defaultRealm;
     private KDCType kdcType;
     private Map<String, String> kerberosEnvProperties;
-
-
-    public void setSecurityEnabled(boolean securityEnabled) {
-      this.securityEnabled = securityEnabled;
-    }
-
-    public boolean isSecurityEnabled() {
-      return securityEnabled;
-    }
+    private SecurityType securityType;
 
     public void setDefaultRealm(String defaultRealm) {
       this.defaultRealm = defaultRealm;
@@ -1547,6 +1533,14 @@ public class KerberosHelper {
     public Map<String, String> getKerberosEnvProperties() {
       return kerberosEnvProperties;
     }
+
+    public void setSecurityType(SecurityType securityType) {
+      this.securityType = securityType;
+    }
+
+    public SecurityType getSecurityType() {
+      return securityType;
+    }
   }
 
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ClusterResourceProvider.java

@@ -50,6 +50,7 @@ import org.apache.ambari.server.orm.entities.HostGroupEntity;
 import org.apache.ambari.server.state.Config;
 import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.ambari.server.state.ConfigImpl;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.kerberos.KerberosDescriptor;
 
@@ -65,6 +66,7 @@ public class ClusterResourceProvider extends BaseBlueprintProcessor {
   public static final String CLUSTER_NAME_PROPERTY_ID    = PropertyHelper.getPropertyId("Clusters", "cluster_name");
   protected static final String CLUSTER_VERSION_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "version");
   protected static final String CLUSTER_PROVISIONING_STATE_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "provisioning_state");
+  protected static final String CLUSTER_SECURITY_TYPE_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "security_type");
   protected static final String CLUSTER_DESIRED_CONFIGS_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "desired_configs");
   protected static final String CLUSTER_DESIRED_SERVICE_CONFIG_VERSIONS_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "desired_service_config_versions");
   protected static final String CLUSTER_TOTAL_HOSTS_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "total_hosts");
@@ -105,6 +107,7 @@ public class ClusterResourceProvider extends BaseBlueprintProcessor {
     propertyIds.add(CLUSTER_NAME_PROPERTY_ID);
     propertyIds.add(CLUSTER_VERSION_PROPERTY_ID);
     propertyIds.add(CLUSTER_PROVISIONING_STATE_PROPERTY_ID);
+    propertyIds.add(CLUSTER_SECURITY_TYPE_PROPERTY_ID);
     propertyIds.add(CLUSTER_DESIRED_CONFIGS_PROPERTY_ID);
     propertyIds.add(CLUSTER_DESIRED_SERVICE_CONFIG_VERSIONS_PROPERTY_ID);
     propertyIds.add(CLUSTER_TOTAL_HOSTS_PROPERTY_ID);
@@ -200,6 +203,7 @@ public class ClusterResourceProvider extends BaseBlueprintProcessor {
       setResourceProperty(resource, CLUSTER_ID_PROPERTY_ID, response.getClusterId(), requestedIds);
       setResourceProperty(resource, CLUSTER_NAME_PROPERTY_ID, clusterName, requestedIds);
       setResourceProperty(resource, CLUSTER_PROVISIONING_STATE_PROPERTY_ID, response.getProvisioningState(), requestedIds);
+      setResourceProperty(resource, CLUSTER_SECURITY_TYPE_PROPERTY_ID, response.getSecurityType(), requestedIds);
       setResourceProperty(resource, CLUSTER_DESIRED_CONFIGS_PROPERTY_ID, response.getDesiredConfigs(), requestedIds);
       setResourceProperty(resource, CLUSTER_DESIRED_SERVICE_CONFIG_VERSIONS_PROPERTY_ID,
         response.getDesiredServiceConfigVersions(), requestedIds);
@@ -366,10 +370,23 @@ public class ClusterResourceProvider extends BaseBlueprintProcessor {
   private ClusterRequest getRequest(Map<String, Object> properties) {
     KerberosDescriptor kerberosDescriptor = new KerberosDescriptor(createKerberosPropertyMap(properties));
 
+    SecurityType securityType;
+    String requestedSecurityType = (String) properties.get(CLUSTER_SECURITY_TYPE_PROPERTY_ID);
+    if(requestedSecurityType == null)
+      securityType = null;
+    else {
+      try {
+        securityType = SecurityType.valueOf(requestedSecurityType.toUpperCase());
+      } catch (IllegalArgumentException e) {
+        throw new IllegalArgumentException(String.format("Cannot set cluster security type to invalid value: %s", requestedSecurityType));
+      }
+    }
+
     ClusterRequest cr = new ClusterRequest(
         (Long) properties.get(CLUSTER_ID_PROPERTY_ID),
         (String) properties.get(CLUSTER_NAME_PROPERTY_ID),
         (String) properties.get(CLUSTER_PROVISIONING_STATE_PROPERTY_ID),
+        securityType,
         (String) properties.get(CLUSTER_VERSION_PROPERTY_ID),
         null,
         kerberosDescriptor,
@@ -904,10 +921,23 @@ public class ClusterResourceProvider extends BaseBlueprintProcessor {
         // only create one cluster request per service, which includes
         // all the configuration types for that service
         if (clusterRequest == null) {
+          SecurityType securityType;
+          String requestedSecurityType = (String) clusterProperties.get(CLUSTER_SECURITY_TYPE_PROPERTY_ID);
+          if(requestedSecurityType == null)
+            securityType = null;
+          else {
+            try {
+              securityType = SecurityType.valueOf(requestedSecurityType.toUpperCase());
+            } catch (IllegalArgumentException e) {
+              throw new IllegalArgumentException(String.format("Cannot set cluster security type to invalid value: %s", requestedSecurityType));
+            }
+          }
+
           clusterRequest = new ClusterRequest(
             (Long) clusterProperties.get(CLUSTER_ID_PROPERTY_ID),
             (String) clusterProperties.get(CLUSTER_NAME_PROPERTY_ID),
             (String) clusterProperties.get(CLUSTER_PROVISIONING_STATE_PROPERTY_ID),
+            securityType,
             (String) clusterProperties.get(CLUSTER_VERSION_PROPERTY_ID),
             null);
         }

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java

@@ -355,7 +355,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
         (String) properties.get(SERVICE_CLUSTER_NAME_PROPERTY_ID),
         (String) properties.get(SERVICE_SERVICE_NAME_PROPERTY_ID),
         (String) properties.get(SERVICE_SERVICE_STATE_PROPERTY_ID));
-    
+
     Object o = properties.get(SERVICE_MAINTENANCE_STATE_PROPERTY_ID);
     if (null != o) {
       svcRequest.setMaintenanceState(o.toString());
@@ -678,7 +678,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
             " in the maintenance state of " + s.getMaintenanceState());
         continue;
       }
-      
+
       seenNewStates.add(newState);
 
       if (newState != oldState) {
@@ -878,19 +878,19 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
       throws AmbariException {
 
     Clusters clusters    = getManagementController().getClusters();
-    
+
     Set<Service> removable = new HashSet<Service>();
-    
+
     for (ServiceRequest serviceRequest : request) {
       if (StringUtils.isEmpty(serviceRequest.getClusterName()) || StringUtils.isEmpty(serviceRequest.getServiceName())) {
         // FIXME throw correct error
         throw new AmbariException("invalid arguments");
       } else {
-        
+
         Service service = clusters.getCluster(
             serviceRequest.getClusterName()).getService(
                 serviceRequest.getServiceName());
-        
+
         if (!service.getDesiredState().isRemovableState()) {
           throw new AmbariException("Cannot remove " + service.getName() + ". Desired state " +
               service.getDesiredState() + " is not removable.  Service must be stopped or disabled.");
@@ -903,15 +903,15 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
             }
           }
         }
-        
+
         removable.add(service);
       }
     }
-    
+
     for (Service service : removable) {
       service.getCluster().deleteService(service.getName());
     }
-    
+
     return null;
   }
 
@@ -1368,7 +1368,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
 
             Set<ServiceComponentHostResponse> hostComponentResponses =
                 controller.getHostComponents(Collections.singleton(request));
-            
+
             State state = State.UNKNOWN;
             for (ServiceComponentHostResponse schr : hostComponentResponses) {
               State schState = getHostComponentState(schr);
@@ -1382,7 +1382,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
           LOG.error("Can't determine service state.", e);
         }
       }
-       
+
       return State.UNKNOWN;
     }
   }

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ClusterEntity.java

@@ -43,6 +43,7 @@ import javax.persistence.OneToOne;
 import javax.persistence.Table;
 import javax.persistence.TableGenerator;
 
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.State;
 
 @Table(name = "clusters")
@@ -83,6 +84,11 @@ public class ClusterEntity {
   @Column(name = "provisioning_state", insertable = true, updatable = true)
   private State provisioningState = State.INIT;
 
+  @Basic
+  @Enumerated(value = EnumType.STRING)
+  @Column(name = "security_type", nullable = false, insertable = true, updatable = true)
+  private SecurityType securityType = SecurityType.NONE;
+
   @Basic
   @Column(name = "desired_cluster_state", insertable = true, updatable = true)
   private String desiredClusterState = "";
@@ -193,6 +199,24 @@ public class ClusterEntity {
     this.provisioningState = provisioningState;
   }
 
+  /**
+   * Gets this ClusterEntity's security type.
+   *
+   * @return the current SecurityType
+   */
+  public SecurityType getSecurityType() {
+    return securityType;
+  }
+
+  /**
+   * Set's this ClusterEntity's security type
+   *
+   * @param securityType the new SecurityType
+   */
+  public void setSecurityType(SecurityType securityType) {
+    this.securityType = securityType;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java

@@ -210,6 +210,20 @@ public interface Cluster {
    */
   public void setProvisioningState(State provisioningState);
 
+  /**
+   * Gets the cluster's security type.
+   *
+   * @return this Cluster's security type
+   */
+  public SecurityType getSecurityType();
+
+  /**
+   * Sets this Cluster's security type.
+   *
+   * @param securityType a SecurityType to set
+   */
+  public void setSecurityType(SecurityType securityType);
+
   /**
    * Gets all configs that match the specified type.  Result is not the
    * DESIRED configuration for a cluster.

ambari-server/src/main/java/org/apache/ambari/server/state/SecurityType.java

@@ -0,0 +1,33 @@
+/*
+ * 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.state;
+
+/**
+ * SecurityType enumerates the different security types Ambari supports.
+ */
+public enum SecurityType {
+  /**
+   * No security.
+   */
+  NONE,
+  /**
+   * Kerberos security.
+   */
+  KERBEROS
+}

ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java

@@ -96,6 +96,7 @@ import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostHealthStatus;
 import org.apache.ambari.server.state.MaintenanceState;
 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;
@@ -1113,6 +1114,44 @@ public class ClusterImpl implements Cluster {
     }
   }
 
+  @Override
+  public SecurityType getSecurityType() {
+    clusterGlobalLock.readLock().lock();
+    try {
+      readLock.lock();
+      SecurityType  securityType = null;
+      try {
+        securityType = clusterEntity.getSecurityType();
+
+        if( null == securityType ) {
+          securityType = SecurityType.NONE;
+        }
+
+        return securityType;
+      } finally {
+        readLock.unlock();
+      }
+    } finally {
+      clusterGlobalLock.readLock().unlock();
+    }
+  }
+
+  @Override
+  public void setSecurityType(SecurityType securityType) {
+    clusterGlobalLock.readLock().lock();
+    try {
+      writeLock.lock();
+      try {
+        clusterEntity.setSecurityType(securityType);
+        clusterDAO.merge(clusterEntity);
+      } finally {
+        writeLock.unlock();
+      }
+    } finally {
+      clusterGlobalLock.readLock().unlock();
+    }
+  }
+
   /**
    * Get the ClusterVersionEntity object whose state is CURRENT.
    * @return
@@ -1693,7 +1732,7 @@ public class ClusterImpl implements Cluster {
         Map<String, Host> hosts = clusters.getHostsForCluster(getClusterName());
 
         return new ClusterResponse(getClusterId(),
-          getClusterName(), getProvisioningState(), hosts.keySet(), hosts.size(),
+          getClusterName(), getProvisioningState(), getSecurityType(), hosts.keySet(), hosts.size(),
           getDesiredStackVersion().getStackId(), getClusterHealthReport());
       } finally {
         readWriteLock.readLock().unlock();

ambari-server/src/main/java/org/apache/ambari/server/upgrade/AbstractUpgradeCatalog.java

@@ -395,7 +395,7 @@ public abstract class AbstractUpgradeCatalog implements UpgradeCatalog {
 
   @Override
   public String toString() {
-    return "{ ugradeCatalog: sourceVersion = " + getSourceVersion() + ", " +
+    return "{ upgradeCatalog: sourceVersion = " + getSourceVersion() + ", " +
       "targetVersion = " + getTargetVersion() + " }";
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog200.java

@@ -45,7 +45,9 @@ import org.apache.ambari.server.orm.entities.ServiceComponentDesiredStateEntityP
 import org.apache.ambari.server.orm.entities.ServiceDesiredStateEntity;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.Config;
 import org.apache.ambari.server.state.SecurityState;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.UpgradeState;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -110,6 +112,10 @@ public class UpgradeCatalog200 extends AbstractUpgradeCatalog {
     executeAlertDDLUpdates();
     createArtifactTable();
 
+    // add security_type to clusters
+    dbAccessor.addColumn("clusters", new DBColumnInfo(
+        "security_type", String.class, 32, SecurityType.NONE.toString(), false));
+
     // add security_state to various tables
     dbAccessor.addColumn("hostcomponentdesiredstate", new DBColumnInfo(
         "security_state", String.class, 32, SecurityState.UNSECURED.toString(), false));
@@ -276,6 +282,7 @@ public class UpgradeCatalog200 extends AbstractUpgradeCatalog {
     removeNagiosService();
     addNewConfigurationsFromXml();
     updateHiveDatabaseType();
+    setSecurityType();
   }
 
   protected void updateHiveDatabaseType() throws AmbariException {
@@ -320,6 +327,43 @@ public class UpgradeCatalog200 extends AbstractUpgradeCatalog {
     executeInTransaction(new RemoveNagiosRunnable());
   }
 
+  /**
+   * Processes existing clusters to set it's security type as indicated by it's <code>cluster-env/security_enabled</code> flag.
+   * <p/>
+   * If the <code>cluster-env/security_enabled</code is set to "true", the cluster's security state
+   * will be set to "KERBEROS" since that is the only option. Else, the value will be set to  "NONE".
+   */
+  protected void setSecurityType() {
+    AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class);
+    Clusters clusters = ambariManagementController.getClusters();
+
+    if (clusters != null) {
+      Map<String, Cluster> clusterMap = clusters.getClusters();
+
+      if (clusterMap != null) {
+        for (final Cluster cluster : clusterMap.values()) {
+          Config configClusterEnv = cluster.getDesiredConfigByType("cluster-env");
+
+          if (configClusterEnv != null) {
+            Map<String, String> properties = configClusterEnv.getProperties();
+
+            if (properties != null) {
+              String securityEnabled = properties.get("security_enabled");
+
+              if ("true".equalsIgnoreCase(securityEnabled)) {
+                // Currently the only security option is Kerberos. If security is enabled, blindly
+                // set security type to SecurityType.KERBEROS
+                cluster.setSecurityType(SecurityType.KERBEROS);
+              } else {
+                cluster.setSecurityType(SecurityType.NONE);
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
   /**
    * The RemoveNagiosRunnable is used to remove Nagios from the cluster. This
    * runnable is exepected to run inside of a transation so that if any of the

ambari-server/src/main/resources/Ambari-DDL-MySQL-CREATE.sql

@@ -32,6 +32,7 @@ CREATE TABLE clusters (
   cluster_info VARCHAR(255) NOT NULL,
   cluster_name VARCHAR(100) NOT NULL UNIQUE,
   provisioning_state VARCHAR(255) NOT NULL DEFAULT 'INIT',
+  security_type VARCHAR(32) NOT NULL DEFAULT 'NONE',
   desired_cluster_state VARCHAR(255) NOT NULL,
   desired_stack_version VARCHAR(255) NOT NULL,
   PRIMARY KEY (cluster_id));

ambari-server/src/main/resources/Ambari-DDL-Oracle-CREATE.sql

@@ -23,6 +23,7 @@ CREATE TABLE clusters (
   cluster_info VARCHAR2(255) NULL,
   cluster_name VARCHAR2(100) NOT NULL UNIQUE,
   provisioning_state VARCHAR2(255) DEFAULT 'INIT' NOT NULL,
+  security_type VARCHAR2(32) DEFAULT 'NONE' NOT NULL,
   desired_cluster_state VARCHAR2(255) NULL,
   desired_stack_version VARCHAR2(255) NULL,
   PRIMARY KEY (cluster_id));

ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql

@@ -23,6 +23,7 @@ CREATE TABLE clusters (
   cluster_info VARCHAR(255) NOT NULL,
   cluster_name VARCHAR(100) NOT NULL UNIQUE,
   provisioning_state VARCHAR(255) NOT NULL DEFAULT 'INIT',
+  security_type VARCHAR(32) NOT NULL DEFAULT 'NONE',
   desired_cluster_state VARCHAR(255) NOT NULL,
   desired_stack_version VARCHAR(255) NOT NULL,
   PRIMARY KEY (cluster_id));

ambari-server/src/main/resources/Ambari-DDL-Postgres-EMBEDDED-CREATE.sql

@@ -34,6 +34,7 @@ CREATE TABLE ambari.clusters (
   cluster_info VARCHAR(255) NOT NULL,
   cluster_name VARCHAR(100) NOT NULL UNIQUE,
   provisioning_state VARCHAR(255) NOT NULL DEFAULT 'INIT',
+  security_type VARCHAR(32) NOT NULL DEFAULT 'NONE',
   desired_cluster_state VARCHAR(255) NOT NULL,
   desired_stack_version VARCHAR(255) NOT NULL,
   PRIMARY KEY (cluster_id));

ambari-server/src/main/resources/Ambari-DDL-SQLServer-CREATE.sql

@@ -31,7 +31,7 @@ GO
 ------create the database------
 
 ------create tables and grant privileges to db user---------
-CREATE TABLE clusters (cluster_id BIGINT NOT NULL, resource_id BIGINT NOT NULL, cluster_info VARCHAR(255) NOT NULL, cluster_name VARCHAR(100) NOT NULL UNIQUE, provisioning_state VARCHAR(255) NOT NULL DEFAULT 'INIT', desired_cluster_state VARCHAR(255) NOT NULL, desired_stack_version VARCHAR(255) NOT NULL, PRIMARY KEY CLUSTERED (cluster_id));
+CREATE TABLE clusters (cluster_id BIGINT NOT NULL, resource_id BIGINT NOT NULL, cluster_info VARCHAR(255) NOT NULL, cluster_name VARCHAR(100) NOT NULL UNIQUE, provisioning_state VARCHAR(255) NOT NULL DEFAULT 'INIT', security_type VARCHAR(32) NOT NULL DEFAULT 'NONE', desired_cluster_state VARCHAR(255) NOT NULL, desired_stack_version VARCHAR(255) NOT NULL, PRIMARY KEY CLUSTERED (cluster_id));
 CREATE TABLE clusterconfig (config_id BIGINT NOT NULL, version_tag VARCHAR(255) NOT NULL, version BIGINT NOT NULL, type_name VARCHAR(255) NOT NULL, cluster_id BIGINT NOT NULL, config_data VARCHAR(MAX) NOT NULL, config_attributes VARCHAR(MAX), create_timestamp BIGINT NOT NULL, PRIMARY KEY CLUSTERED (config_id));
 CREATE TABLE serviceconfig (service_config_id BIGINT NOT NULL, cluster_id BIGINT NOT NULL, service_name VARCHAR(255) NOT NULL, version BIGINT NOT NULL, create_timestamp BIGINT NOT NULL, user_name VARCHAR(255) NOT NULL DEFAULT '_db', group_id BIGINT, note VARCHAR(MAX), PRIMARY KEY CLUSTERED (service_config_id));
 CREATE TABLE serviceconfighosts (service_config_id BIGINT NOT NULL, hostname VARCHAR(255) NOT NULL, PRIMARY KEY CLUSTERED (service_config_id, hostname));

ambari-server/src/main/resources/properties.json

@@ -3,6 +3,7 @@
         "Clusters/cluster_id",
         "Clusters/cluster_name",
         "Clusters/provisioning_state",
+        "Clusters/security_type",
         "Clusters/version",
         "Clusters/state",
         "Clusters/desired_configs",

ambari-server/src/test/java/org/apache/ambari/server/controller/AmbariCustomCommandExecutionHelperTest.java

@@ -46,6 +46,8 @@ import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
+import org.apache.ambari.server.state.SecurityState;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.State;
 import org.junit.After;
 import org.junit.Before;
@@ -322,7 +324,8 @@ public class
   }
 
   private void createCluster(String clusterName) throws AmbariException {
-    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), "HDP-2.0.6", null);
+    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(),
+        SecurityType.NONE, "HDP-2.0.6", null);
     controller.createCluster(r);
   }
   

ambari-server/src/test/java/org/apache/ambari/server/controller/AmbariManagementControllerImplTest.java

@@ -36,6 +36,7 @@ import org.apache.ambari.server.actionmanager.ActionDBAccessorImpl;
 import org.apache.ambari.server.actionmanager.ActionManager;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
 import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.controller.internal.RequestStageContainer;
 import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.orm.entities.LdapSyncSpecEntity;
@@ -47,13 +48,14 @@ import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.ComponentInfo;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.MaintenanceState;
+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.ServiceComponentImpl;
 import org.apache.ambari.server.state.ServiceInfo;
 import org.apache.ambari.server.state.ServiceOsSpecific;
 import org.apache.ambari.server.state.StackId;
+import org.apache.ambari.server.state.kerberos.KerberosDescriptor;
 import org.easymock.Capture;
 import org.junit.Before;
 import org.junit.Test;
@@ -68,11 +70,8 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.TreeMap;
 
-import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.DB_DRIVER_FILENAME;
-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.*;
 import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.capture;
 import static org.easymock.EasyMock.createMock;
@@ -538,7 +537,7 @@ public class AmbariManagementControllerImplTest {
 
   /**
    * Ensure that when the cluster is updated KerberosHandler.toggleKerberos is not invoked unless
-   * the security state is altered
+   * the security type is altered
    */
   @Test
   public void testUpdateClustersToggleKerberosNotInvoked() throws Exception {
@@ -566,7 +565,101 @@ public class AmbariManagementControllerImplTest {
     expectLastCall().once();
 
     // replay mocks
-    replay(actionManager, cluster, clusters, injector, clusterRequest, sessionManager);
+    replay(actionManager, cluster, clusters, injector, clusterRequest, sessionManager, kerberosHelper);
+
+    // test
+    AmbariManagementController controller = new AmbariManagementControllerImpl(actionManager, clusters, injector);
+    controller.updateClusters(setRequests, null);
+
+    // assert and verify
+    assertSame(controller, controllerCapture.getValue());
+    verify(actionManager, cluster, clusters, injector, clusterRequest, sessionManager);
+  }
+
+  /**
+   * Ensure that when the cluster security type updated from NONE to KERBEROS, KerberosHandler.toggleKerberos
+   * IS invoked
+   */
+  @Test
+  public void testUpdateClustersToggleKerberosEnable() throws Exception {
+    // member state mocks
+    Capture<AmbariManagementController> controllerCapture = new Capture<AmbariManagementController>();
+    Injector injector = createStrictMock(Injector.class);
+    Cluster cluster = createNiceMock(Cluster.class);
+    ActionManager actionManager = createNiceMock(ActionManager.class);
+    ClusterRequest clusterRequest = createNiceMock(ClusterRequest.class);
+
+    // requests
+    Set<ClusterRequest> setRequests = Collections.singleton(clusterRequest);
+
+    KerberosHelper kerberosHelper = createStrictMock(KerberosHelper.class);
+    // expectations
+    injector.injectMembers(capture(controllerCapture));
+    expect(injector.getInstance(Gson.class)).andReturn(null);
+    expect(injector.getInstance(MaintenanceStateHelper.class)).andReturn(null);
+    expect(injector.getInstance(KerberosHelper.class)).andReturn(kerberosHelper);
+    expect(clusterRequest.getClusterId()).andReturn(1L).times(6);
+    expect(clusterRequest.getSecurityType()).andReturn(SecurityType.KERBEROS).anyTimes();
+    expect(clusters.getClusterById(1L)).andReturn(cluster).times(2);
+    expect(cluster.getClusterName()).andReturn("cluster").times(2);
+    expect(cluster.getSecurityType()).andReturn(SecurityType.NONE).anyTimes();
+
+    cluster.addSessionAttributes(anyObject(Map.class));
+    expectLastCall().once();
+
+    expect(kerberosHelper.toggleKerberos(anyObject(Cluster.class), anyObject(SecurityType.class), anyObject(KerberosDescriptor.class), anyObject(RequestStageContainer.class)))
+        .andReturn(null)
+        .once();
+
+    // replay mocks
+    replay(actionManager, cluster, clusters, injector, clusterRequest, sessionManager, kerberosHelper);
+
+    // test
+    AmbariManagementController controller = new AmbariManagementControllerImpl(actionManager, clusters, injector);
+    controller.updateClusters(setRequests, null);
+
+    // assert and verify
+    assertSame(controller, controllerCapture.getValue());
+    verify(actionManager, cluster, clusters, injector, clusterRequest, sessionManager);
+  }
+
+  /**
+   * Ensure that when the cluster security type updated from KERBEROS to NONE, KerberosHandler.toggleKerberos
+   * IS invoked
+   */
+  @Test
+  public void testUpdateClustersToggleKerberosDisable() throws Exception {
+    // member state mocks
+    Capture<AmbariManagementController> controllerCapture = new Capture<AmbariManagementController>();
+    Injector injector = createStrictMock(Injector.class);
+    Cluster cluster = createNiceMock(Cluster.class);
+    ActionManager actionManager = createNiceMock(ActionManager.class);
+    ClusterRequest clusterRequest = createNiceMock(ClusterRequest.class);
+
+    // requests
+    Set<ClusterRequest> setRequests = Collections.singleton(clusterRequest);
+
+    KerberosHelper kerberosHelper = createStrictMock(KerberosHelper.class);
+    // expectations
+    injector.injectMembers(capture(controllerCapture));
+    expect(injector.getInstance(Gson.class)).andReturn(null);
+    expect(injector.getInstance(MaintenanceStateHelper.class)).andReturn(null);
+    expect(injector.getInstance(KerberosHelper.class)).andReturn(kerberosHelper);
+    expect(clusterRequest.getClusterId()).andReturn(1L).times(6);
+    expect(clusterRequest.getSecurityType()).andReturn(SecurityType.NONE).anyTimes();
+    expect(clusters.getClusterById(1L)).andReturn(cluster).times(2);
+    expect(cluster.getClusterName()).andReturn("cluster").times(2);
+    expect(cluster.getSecurityType()).andReturn(SecurityType.KERBEROS).anyTimes();
+
+    cluster.addSessionAttributes(anyObject(Map.class));
+    expectLastCall().once();
+
+    expect(kerberosHelper.toggleKerberos(anyObject(Cluster.class), anyObject(SecurityType.class), anyObject(KerberosDescriptor.class), anyObject(RequestStageContainer.class)))
+        .andReturn(null)
+        .once();
+
+    // replay mocks
+    replay(actionManager, cluster, clusters, injector, clusterRequest, sessionManager, kerberosHelper);
 
     // test
     AmbariManagementController controller = new AmbariManagementControllerImpl(actionManager, clusters, injector);
@@ -577,6 +670,69 @@ public class AmbariManagementControllerImplTest {
     verify(actionManager, cluster, clusters, injector, clusterRequest, sessionManager);
   }
 
+  /**
+   * Ensure that when the cluster security type updated from KERBEROS to NONE, KerberosHandler.toggleKerberos
+   * IS invoked
+   */
+  @Test
+  public void testUpdateClustersToggleKerberosFail() throws Exception {
+    // member state mocks
+    Capture<AmbariManagementController> controllerCapture = new Capture<AmbariManagementController>();
+    Injector injector = createStrictMock(Injector.class);
+    Cluster cluster = createMock(Cluster.class);
+    ActionManager actionManager = createNiceMock(ActionManager.class);
+    ClusterRequest clusterRequest = createNiceMock(ClusterRequest.class);
+
+    // requests
+    Set<ClusterRequest> setRequests = Collections.singleton(clusterRequest);
+
+    KerberosHelper kerberosHelper = createStrictMock(KerberosHelper.class);
+    // expectations
+    injector.injectMembers(capture(controllerCapture));
+    expect(injector.getInstance(Gson.class)).andReturn(null);
+    expect(injector.getInstance(MaintenanceStateHelper.class)).andReturn(null);
+    expect(injector.getInstance(KerberosHelper.class)).andReturn(kerberosHelper);
+    expect(clusterRequest.getClusterId()).andReturn(1L).times(6);
+    expect(clusterRequest.getSecurityType()).andReturn(SecurityType.NONE).anyTimes();
+    expect(clusters.getClusterById(1L)).andReturn(cluster).times(2);
+    expect(cluster.getClusterName()).andReturn("cluster").times(2);
+    expect(cluster.getSecurityType()).andReturn(SecurityType.KERBEROS).anyTimes();
+    expect(cluster.getCurrentClusterVersion()).andReturn(null).anyTimes();
+    expect(cluster.getCurrentStackVersion()).andReturn(null).anyTimes();
+    expect(cluster.getDesiredStackVersion()).andReturn(null).anyTimes();
+
+    cluster.setCurrentStackVersion(anyObject(StackId.class));
+    expectLastCall().once();
+
+    cluster.setClusterName(anyObject(String.class));
+    expectLastCall().once();
+
+    cluster.addSessionAttributes(anyObject(Map.class));
+    expectLastCall().once();
+
+    expect(kerberosHelper.toggleKerberos(anyObject(Cluster.class), anyObject(SecurityType.class), anyObject(KerberosDescriptor.class), anyObject(RequestStageContainer.class)))
+        .andThrow(new IllegalArgumentException("bad args!"))
+        .once();
+
+    // replay mocks
+    replay(actionManager, cluster, clusters, injector, clusterRequest, sessionManager, kerberosHelper);
+
+    // test
+    AmbariManagementController controller = new AmbariManagementControllerImpl(actionManager, clusters, injector);
+
+    try {
+      controller.updateClusters(setRequests, null);
+      Assert.fail("IllegalArgumentException not thrown");
+    }
+    catch(IllegalArgumentException e) {
+      // This is expected
+    }
+
+    // assert and verify
+    assertSame(controller, controllerCapture.getValue());
+    verify(actionManager, cluster, clusters, injector, clusterRequest, sessionManager);
+  }
+
   /**
    * Ensure that RollbackException is thrown outside the updateClusters method
    * when a unique constraint violation occurs.

ambari-server/src/test/java/org/apache/ambari/server/controller/AmbariManagementControllerTest.java

@@ -96,29 +96,7 @@ import org.apache.ambari.server.orm.dao.HostDAO;
 import org.apache.ambari.server.orm.entities.ExecutionCommandEntity;
 import org.apache.ambari.server.security.authorization.Users;
 import org.apache.ambari.server.serveraction.ServerAction;
-import org.apache.ambari.server.state.Cluster;
-import org.apache.ambari.server.state.Clusters;
-import org.apache.ambari.server.state.Config;
-import org.apache.ambari.server.state.ConfigFactory;
-import org.apache.ambari.server.state.ConfigHelper;
-import org.apache.ambari.server.state.ConfigImpl;
-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.RepositoryInfo;
-import org.apache.ambari.server.state.RepositoryVersionState;
-import org.apache.ambari.server.state.Service;
-import org.apache.ambari.server.state.ServiceComponent;
-import org.apache.ambari.server.state.ServiceComponentFactory;
-import org.apache.ambari.server.state.ServiceComponentHost;
-import org.apache.ambari.server.state.ServiceComponentHostFactory;
-import org.apache.ambari.server.state.ServiceFactory;
-import org.apache.ambari.server.state.ServiceInfo;
-import org.apache.ambari.server.state.ServiceOsSpecific;
-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.*;
 import org.apache.ambari.server.state.configgroup.ConfigGroup;
 import org.apache.ambari.server.state.configgroup.ConfigGroupFactory;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostInstallEvent;
@@ -258,7 +236,7 @@ public class AmbariManagementControllerTest {
    * @throws AmbariException
    */
   private void createCluster(String clusterName) throws AmbariException {
-    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), "HDP-0.1", null);
+    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), SecurityType.NONE, "HDP-0.1", null);
     controller.createCluster(r);
   }
 

ambari-server/src/test/java/org/apache/ambari/server/controller/BackgroundCustomCommandExecutionTest.java

@@ -45,6 +45,7 @@ import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.State;
 import org.junit.After;
 import org.junit.Before;
@@ -178,7 +179,7 @@ public class BackgroundCustomCommandExecutionTest {
   }
 
   private void createCluster(String clusterName) throws AmbariException {
-    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), "HDP-2.0.6", null);
+    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), SecurityType.NONE, "HDP-2.0.6", null);
     controller.createCluster(r);
   }
   

ambari-server/src/test/java/org/apache/ambari/server/controller/ClusterRequestTest.java

@@ -21,6 +21,8 @@ package org.apache.ambari.server.controller;
 import java.util.HashSet;
 import java.util.Set;
 
+import org.apache.ambari.server.state.SecurityState;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.State;
 import org.junit.Assert;
@@ -33,17 +35,19 @@ public class ClusterRequestTest {
     Long clusterId = new Long(10);
     String clusterName = "foo";
     String provisioningState = State.INIT.name();
+    SecurityType securityType = SecurityType.NONE;
     StackId stackVersion = new StackId("HDP-1.0.1");
     Set<String> hostNames = new HashSet<String>();
     hostNames.add("h1");
 
     ClusterRequest r1 =
-        new ClusterRequest(clusterId, clusterName, provisioningState,
+        new ClusterRequest(clusterId, clusterName, provisioningState, securityType,
             stackVersion.getStackId(), hostNames);
 
     Assert.assertEquals(clusterId, r1.getClusterId());
     Assert.assertEquals(clusterName, r1.getClusterName());
     Assert.assertEquals(provisioningState, r1.getProvisioningState());
+    Assert.assertEquals(securityType, r1.getSecurityType());
     Assert.assertEquals(stackVersion.getStackId(),
         r1.getStackVersion());
     Assert.assertArrayEquals(hostNames.toArray(), r1.getHostNames().toArray());
@@ -53,19 +57,21 @@ public class ClusterRequestTest {
     r1.setClusterName("foo1");
     r1.setStackVersion("HDP-1.0.2");
     r1.setProvisioningState(State.INSTALLED.name());
+    r1.setSecurityType(SecurityType.KERBEROS);
 
     hostNames.add("h2");
 
     Assert.assertEquals(clusterId, r1.getClusterId());
     Assert.assertEquals("foo1", r1.getClusterName());
     Assert.assertEquals(State.INSTALLED.name(), r1.getProvisioningState());
+    Assert.assertEquals(SecurityType.KERBEROS, r1.getSecurityType());
     Assert.assertEquals("HDP-1.0.2", r1.getStackVersion());
     Assert.assertArrayEquals(hostNames.toArray(), r1.getHostNames().toArray());
   }
 
   @Test
   public void testToString() {
-    ClusterRequest r1 = new ClusterRequest(null, null, null, null, null);
+    ClusterRequest r1 = new ClusterRequest(null, null, null, null, null, null);
     r1.toString();
   }
 

ambari-server/src/test/java/org/apache/ambari/server/controller/ClusterResponseTest.java

@@ -21,6 +21,7 @@ package org.apache.ambari.server.controller;
 import java.util.HashSet;
 import java.util.Set;
 
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.State;
 import org.junit.Assert;
 import org.junit.Test;
@@ -32,16 +33,18 @@ public class ClusterResponseTest {
     Long clusterId = new Long(10);
     String clusterName = "foo";
     State provisioningState = State.INSTALLED;
+    SecurityType securityType = SecurityType.KERBEROS;
     Set<String> hostNames = new HashSet<String>();
     hostNames.add("h1");
 
     ClusterResponse r1 =
-        new ClusterResponse(clusterId, clusterName, provisioningState,
+        new ClusterResponse(clusterId, clusterName, provisioningState, securityType,
           hostNames, hostNames.size(), "bar", null);
     
     Assert.assertEquals(clusterId, r1.getClusterId());
     Assert.assertEquals(clusterName, r1.getClusterName());
     Assert.assertEquals(provisioningState.name(), r1.getProvisioningState());
+    Assert.assertEquals(securityType.name(), r1.getSecurityType());
     Assert.assertArrayEquals(hostNames.toArray(), r1.getHostNames().toArray());
     Assert.assertEquals(Integer.valueOf(1), r1.getTotalHosts());
     Assert.assertEquals("bar", r1.getDesiredStackVersion());
@@ -50,7 +53,7 @@ public class ClusterResponseTest {
   @Test
   public void testToString() {
     ClusterResponse r =
-      new ClusterResponse(null, null, null, null, null, null, null);
+      new ClusterResponse(null, null, null, null, null, null, null, null);
     r.toString();
   }
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/KerberosHelperTest.java

@@ -46,6 +46,7 @@ import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
 import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.SecurityState;
+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;
@@ -161,7 +162,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     RequestStageContainer requestStageContainer = createNiceMock(RequestStageContainer.class);
 
     replayAll();
-    kerberosHelper.toggleKerberos(cluster, kerberosDescriptor, requestStageContainer);
+    kerberosHelper.toggleKerberos(cluster, SecurityType.KERBEROS, kerberosDescriptor, requestStageContainer);
     verifyAll();
   }
 
@@ -169,27 +170,20 @@ public class KerberosHelperTest extends EasyMockSupport {
   public void testMissingKrb5Conf() throws Exception {
     KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
 
-    final Map<String, String> clusterEnvProperties = createNiceMock(Map.class);
-    expect(clusterEnvProperties.get("security_enabled")).andReturn("true").once();
-
-    final Config clusterEnvConfig = createNiceMock(Config.class);
-    expect(clusterEnvConfig.getProperties()).andReturn(clusterEnvProperties).once();
-
     final Map<String, String> kerberosEnvProperties = createNiceMock(Map.class);
-    expect(clusterEnvProperties.get("ldap_url")).andReturn("").once();
-    expect(clusterEnvProperties.get("container_dn")).andReturn("").once();
+    expect(kerberosEnvProperties.get("ldap_url")).andReturn("").once();
+    expect(kerberosEnvProperties.get("container_dn")).andReturn("").once();
 
     final Config kerberosEnvConfig = createNiceMock(Config.class);
     expect(kerberosEnvConfig.getProperties()).andReturn(kerberosEnvProperties).once();
 
     final Cluster cluster = createNiceMock(Cluster.class);
-    expect(cluster.getDesiredConfigByType("cluster-env")).andReturn(clusterEnvConfig).once();
     expect(cluster.getDesiredConfigByType("kerberos-env")).andReturn(kerberosEnvConfig).once();
 
     final KerberosDescriptor kerberosDescriptor = createNiceMock(KerberosDescriptor.class);
 
     replayAll();
-    kerberosHelper.toggleKerberos(cluster, kerberosDescriptor, null);
+    kerberosHelper.toggleKerberos(cluster, SecurityType.KERBEROS, kerberosDescriptor, null);
     verifyAll();
   }
 
@@ -197,12 +191,6 @@ public class KerberosHelperTest extends EasyMockSupport {
   public void testMissingKerberosEnvConf() throws Exception {
     KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
 
-    final Map<String, String> clusterEnvProperties = createNiceMock(Map.class);
-    expect(clusterEnvProperties.get("security_enabled")).andReturn("true").once();
-
-    final Config clusterEnvConfig = createNiceMock(Config.class);
-    expect(clusterEnvConfig.getProperties()).andReturn(clusterEnvProperties).once();
-
     final Map<String, String> krb5ConfProperties = createNiceMock(Map.class);
     expect(krb5ConfProperties.get("kdc_host")).andReturn("10.0.100.1").once();
     expect(krb5ConfProperties.get("kadmin_host")).andReturn("10.0.100.1").once();
@@ -212,13 +200,12 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(krb5ConfConfig.getProperties()).andReturn(krb5ConfProperties).once();
 
     final Cluster cluster = createNiceMock(Cluster.class);
-    expect(cluster.getDesiredConfigByType("cluster-env")).andReturn(clusterEnvConfig).once();
     expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).once();
 
     final KerberosDescriptor kerberosDescriptor = createNiceMock(KerberosDescriptor.class);
 
     replayAll();
-    kerberosHelper.toggleKerberos(cluster, kerberosDescriptor, null);
+    kerberosHelper.toggleKerberos(cluster, SecurityType.KERBEROS, kerberosDescriptor, null);
     verifyAll();
   }
 
@@ -305,12 +292,6 @@ public class KerberosHelperTest extends EasyMockSupport {
     service2.setSecurityState(SecurityState.SECURED_KERBEROS);
     expectLastCall().once();
 
-    final Map<String, String> clusterEnvProperties = createNiceMock(Map.class);
-    expect(clusterEnvProperties.get("security_enabled")).andReturn("true").once();
-
-    final Config clusterEnvConfig = createNiceMock(Config.class);
-    expect(clusterEnvConfig.getProperties()).andReturn(clusterEnvProperties).once();
-
     final Map<String, String> kerberosEnvProperties = createNiceMock(Map.class);
     // TODO: (rlevas) Add when AMBARI 9121 is complete
     // expect(kerberosEnvProperties.get("kdc_type")).andReturn("mit-kdc").once();
@@ -331,7 +312,7 @@ public class KerberosHelperTest extends EasyMockSupport {
         .andReturn(MaintenanceState.OFF).anyTimes();
 
     final Cluster cluster = createNiceMock(Cluster.class);
-    expect(cluster.getDesiredConfigByType("cluster-env")).andReturn(clusterEnvConfig).once();
+    expect(cluster.getSecurityType()).andReturn(SecurityType.KERBEROS).once();
     expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).once();
     expect(cluster.getDesiredConfigByType("kerberos-env")).andReturn(kerberosEnvConfig).once();
     expect(cluster.getClusterName()).andReturn("c1").anyTimes();
@@ -501,7 +482,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     // Needed by infrastructure
     injector.getInstance(AmbariMetaInfo.class).init();
 
-    kerberosHelper.toggleKerberos(cluster, kerberosDescriptor, requestStageContainer);
+    kerberosHelper.toggleKerberos(cluster, SecurityType.KERBEROS, kerberosDescriptor, requestStageContainer);
 
     verifyAll();
   }
@@ -538,12 +519,6 @@ public class KerberosHelperTest extends EasyMockSupport {
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
         .once();
 
-    final Map<String, String> clusterEnvProperties = createNiceMock(Map.class);
-    expect(clusterEnvProperties.get("security_enabled")).andReturn("true").once();
-
-    final Config clusterEnvConfig = createNiceMock(Config.class);
-    expect(clusterEnvConfig.getProperties()).andReturn(clusterEnvProperties).once();
-
     final Map<String, String> kerberosEnvProperties = createNiceMock(Map.class);
     // TODO: (rlevas) Add when AMBARI 9121 is complete
     // expect(kerberosEnvProperties.get("kdc_type")).andReturn("mit-kdc").once();
@@ -564,7 +539,6 @@ public class KerberosHelperTest extends EasyMockSupport {
         .andReturn(MaintenanceState.OFF).anyTimes();
 
     final Cluster cluster = createNiceMock(Cluster.class);
-    expect(cluster.getDesiredConfigByType("cluster-env")).andReturn(clusterEnvConfig).once();
     expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).once();
     expect(cluster.getDesiredConfigByType("kerberos-env")).andReturn(kerberosEnvConfig).once();
     expect(cluster.getClusterName()).andReturn("c1").anyTimes();

ambari-server/src/test/java/org/apache/ambari/server/controller/RefreshYarnCapacitySchedulerReleaseConfigTest.java

@@ -37,6 +37,7 @@ import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
+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;
@@ -182,7 +183,7 @@ public class RefreshYarnCapacitySchedulerReleaseConfigTest {
   }
 
   private void createCluster(String clusterName, String stackName) throws AmbariException {
-    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), stackName, null);
+    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), SecurityType.NONE, stackName, null);
     controller.createCluster(r);
   }
   

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/AbstractResourceProviderTest.java

@@ -53,6 +53,7 @@ import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
 import org.apache.ambari.server.controller.spi.SystemException;
 import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
 import org.apache.ambari.server.controller.utilities.PredicateBuilder;
+import org.apache.ambari.server.state.SecurityType;
 import org.easymock.EasyMock;
 import org.easymock.IArgumentMatcher;
 import org.junit.Assert;
@@ -304,9 +305,9 @@ public class AbstractResourceProviderTest {
     }
 
     public static Set<ClusterRequest> getClusterRequestSet(
-        Long clusterId, String clusterName, String provisioningState, String stackVersion, Set<String> hostNames)
+        Long clusterId, String clusterName, String provisioningState, SecurityType securityType, String stackVersion, Set<String> hostNames)
     {
-      EasyMock.reportMatcher(new ClusterRequestSetMatcher(clusterId, clusterName, provisioningState, stackVersion, hostNames));
+      EasyMock.reportMatcher(new ClusterRequestSetMatcher(clusterId, clusterName, provisioningState, securityType, stackVersion, hostNames));
       return null;
     }
 
@@ -401,8 +402,9 @@ public class AbstractResourceProviderTest {
    */
   public static class ClusterRequestSetMatcher extends ClusterRequest implements IArgumentMatcher {
 
-    public ClusterRequestSetMatcher(Long clusterId, String clusterName, String provisioningState, String stackVersion, Set<String> hostNames) {
-      super(clusterId, clusterName, provisioningState, stackVersion, hostNames);
+    public ClusterRequestSetMatcher(Long clusterId, String clusterName, String provisioningState,
+                                    SecurityType securityType, String stackVersion, Set<String> hostNames) {
+      super(clusterId, clusterName, provisioningState, securityType, stackVersion, hostNames);
     }
 
     @Override

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ClusterResourceProviderTest.java

@@ -76,6 +76,7 @@ import org.apache.ambari.server.state.AutoDeployInfo;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.ambari.server.state.DependencyInfo;
+import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.State;
 import org.easymock.Capture;
 import org.easymock.EasyMock;
@@ -2646,17 +2647,17 @@ public class ClusterResourceProviderTest {
     Clusters clusters = createMock(Clusters.class);
 
     Set<ClusterResponse> allResponse = new HashSet<ClusterResponse>();
-    allResponse.add(new ClusterResponse(100L, "Cluster100", State.INSTALLED, null, null, null, null));
-    allResponse.add(new ClusterResponse(101L, "Cluster101", State.INSTALLED, null, null, null, null));
-    allResponse.add(new ClusterResponse(102L, "Cluster102", State.INSTALLED, null, null, null, null));
-    allResponse.add(new ClusterResponse(103L, "Cluster103", State.INSTALLED, null, null, null, null));
-    allResponse.add(new ClusterResponse(104L, "Cluster104", State.INSTALLED, null, null, null, null));
+    allResponse.add(new ClusterResponse(100L, "Cluster100", State.INSTALLED, SecurityType.NONE, null, null, null, null));
+    allResponse.add(new ClusterResponse(101L, "Cluster101", State.INSTALLED, SecurityType.NONE, null, null, null, null));
+    allResponse.add(new ClusterResponse(102L, "Cluster102", State.INSTALLED, SecurityType.NONE, null, null, null, null));
+    allResponse.add(new ClusterResponse(103L, "Cluster103", State.INSTALLED, SecurityType.NONE, null, null, null, null));
+    allResponse.add(new ClusterResponse(104L, "Cluster104", State.INSTALLED, SecurityType.NONE, null, null, null, null));
 
     Set<ClusterResponse> nameResponse = new HashSet<ClusterResponse>();
-    nameResponse.add(new ClusterResponse(102L, "Cluster102", State.INSTALLED, null, null, null, null));
+    nameResponse.add(new ClusterResponse(102L, "Cluster102", State.INSTALLED, SecurityType.NONE, null, null, null, null));
 
     Set<ClusterResponse> idResponse = new HashSet<ClusterResponse>();
-    idResponse.add(new ClusterResponse(103L, "Cluster103", State.INSTALLED, null, null, null, null));
+    idResponse.add(new ClusterResponse(103L, "Cluster103", State.INSTALLED, SecurityType.NONE, null, null, null, null));
 
     // set expectations
     Capture<Set<ClusterRequest>> captureClusterRequests = new Capture<Set<ClusterRequest>>();
@@ -2737,7 +2738,7 @@ public class ClusterResourceProviderTest {
     RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
 
     Set<ClusterResponse> nameResponse = new HashSet<ClusterResponse>();
-    nameResponse.add(new ClusterResponse(102L, "Cluster102", State.INIT, null, null, null, null));
+    nameResponse.add(new ClusterResponse(102L, "Cluster102", State.INIT, SecurityType.NONE, null, null, null, null));
 
     Map<String, String> mapRequestProps = new HashMap<String, String>();
     mapRequestProps.put("context", "Called from a test");
@@ -2745,11 +2746,11 @@ public class ClusterResourceProviderTest {
     // set expectations
     expect(managementController.getClusters(EasyMock.<Set<ClusterRequest>>anyObject())).andReturn(nameResponse).once();
     expect(managementController.updateClusters(
-        AbstractResourceProviderTest.Matcher.getClusterRequestSet(102L, "Cluster102", State.INSTALLED.name(), "HDP-0.1", null), eq(mapRequestProps))).
+        AbstractResourceProviderTest.Matcher.getClusterRequestSet(102L, "Cluster102", State.INSTALLED.name(), SecurityType.NONE, "HDP-0.1", null), eq(mapRequestProps))).
         andReturn(response).once();
     
     expect(managementController.updateClusters(
-        AbstractResourceProviderTest.Matcher.getClusterRequestSet(103L, null, null, "HDP-0.1", null), eq(mapRequestProps))).
+        AbstractResourceProviderTest.Matcher.getClusterRequestSet(103L, null, null, null, "HDP-0.1", null), eq(mapRequestProps))).
         andReturn(response).once();
 
     expect(managementController.getClusterUpdateResults(anyObject(ClusterRequest.class))).andReturn(null).anyTimes();
@@ -2810,7 +2811,7 @@ public class ClusterResourceProviderTest {
     RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
 
     Set<ClusterResponse> nameResponse = new HashSet<ClusterResponse>();
-    nameResponse.add(new ClusterResponse(100L, "Cluster100", State.INSTALLED, null, null, null, null));
+    nameResponse.add(new ClusterResponse(100L, "Cluster100", State.INSTALLED, SecurityType.NONE, null, null, null, null));
 
     Map<String, String> mapRequestProps = new HashMap<String, String>();
     mapRequestProps.put("context", "Called from a test");

ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog200Test.java

@@ -62,6 +62,7 @@ import org.apache.ambari.server.orm.entities.HostEntity;
 import org.apache.ambari.server.orm.entities.ServiceComponentDesiredStateEntity;
 import org.apache.ambari.server.orm.entities.ServiceComponentDesiredStateEntityPK;
 import org.apache.ambari.server.state.SecurityState;
+import org.apache.ambari.server.state.SecurityType;
 import org.easymock.Capture;
 import org.junit.After;
 import org.junit.Assert;
@@ -117,6 +118,7 @@ public class UpgradeCatalog200Test {
     Capture<DBAccessor.DBColumnInfo> alertTargetGlobalColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
     Capture<DBAccessor.DBColumnInfo> hostComponentStateColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
     Capture<DBAccessor.DBColumnInfo> hostComponentVersionColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
+    Capture<DBAccessor.DBColumnInfo> clustersSecurityTypeColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
     Capture<DBAccessor.DBColumnInfo> hostComponentStateSecurityStateColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
     Capture<DBAccessor.DBColumnInfo> hostComponentDesiredStateSecurityStateColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
     Capture<DBAccessor.DBColumnInfo> hostRoleCommandRetryColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
@@ -168,6 +170,10 @@ public class UpgradeCatalog200Test {
     dbAccessor.addColumn(eq("stage"),
         capture(stageSkippableColumnCapture));
 
+    // Clusters: security type
+    dbAccessor.addColumn(eq("clusters"),
+        capture(clustersSecurityTypeColumnCapture));
+
     // Host Component State: security State
     dbAccessor.addColumn(eq("hostcomponentstate"),
         capture(hostComponentStateSecurityStateColumnCapture));
@@ -261,6 +267,9 @@ public class UpgradeCatalog200Test {
     assertEquals(0, upgradeSkippableColumn.getDefaultValue());
     assertFalse(upgradeSkippableColumn.isNullable());
 
+    // verify security_type column
+    verifyClustersSecurityType(clustersSecurityTypeColumnCapture);
+
     // verify security_state columns
     verifyComponentSecurityStateColumn(hostComponentStateSecurityStateColumnCapture);
     verifyComponentSecurityStateColumn(hostComponentDesiredStateSecurityStateColumnCapture);
@@ -297,9 +306,14 @@ public class UpgradeCatalog200Test {
     Method updateHiveDatabaseType = UpgradeCatalog200.class.getDeclaredMethod("updateHiveDatabaseType");
     Method addNewConfigurationsFromXml = AbstractUpgradeCatalog.class.getDeclaredMethod
         ("addNewConfigurationsFromXml");
+    Method setSecurityType = UpgradeCatalog200.class.getDeclaredMethod("setSecurityType");
 
-    UpgradeCatalog200 upgradeCatalog = createMockBuilder(
-        UpgradeCatalog200.class).addMockedMethod(removeNagiosService).addMockedMethod(updateHiveDatabaseType).addMockedMethod(addNewConfigurationsFromXml).createMock();
+    UpgradeCatalog200 upgradeCatalog = createMockBuilder(UpgradeCatalog200.class)
+        .addMockedMethod(removeNagiosService)
+        .addMockedMethod(updateHiveDatabaseType)
+        .addMockedMethod(addNewConfigurationsFromXml)
+        .addMockedMethod(setSecurityType)
+        .createMock();
 
     upgradeCatalog.removeNagiosService();
     expectLastCall().once();
@@ -307,7 +321,8 @@ public class UpgradeCatalog200Test {
     expectLastCall();
     upgradeCatalog.updateHiveDatabaseType();
     expectLastCall().once();
-
+    upgradeCatalog.setSecurityType();
+    expectLastCall().once();
 
     replay(upgradeCatalog);
 
@@ -456,6 +471,20 @@ public class UpgradeCatalog200Test {
     Assert.assertEquals("security_state", column.getName());
   }
 
+  /**
+   * Verifies new security_type column in clusters table
+   *
+   * @param securityTypeColumnCapture
+   */
+  private void verifyClustersSecurityType(
+      Capture<DBAccessor.DBColumnInfo> securityTypeColumnCapture) {
+    DBColumnInfo column = securityTypeColumnCapture.getValue();
+    Assert.assertEquals(SecurityType.NONE.toString(), column.getDefaultValue());
+    Assert.assertEquals(Integer.valueOf(32), column.getLength());
+    Assert.assertEquals(String.class, column.getType());
+    Assert.assertEquals("security_type", column.getName());
+  }
+
   /**
    * Verifies new security_state column in hostcomponentdesiredstate and hostcomponentstate tables
    *