AMBARI-9666. Kerberos: Adding a service to a Kerberized cluster requires Kerberos-related tasks occur before INSTALL stage (rlevas)

ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java

@@ -459,21 +459,22 @@ public class HeartBeatHandler {
           "SET_KEYTAB".equalsIgnoreCase(report.getCustomCommand()) &&
           RequestExecution.Status.COMPLETED.name().equalsIgnoreCase(report.getStatus())) {
 
-        Map<String, String> structuredOutput;
+        WriteKeytabsStructuredOut writeKeytabsStructuredOut;
         try {
-          structuredOutput = gson.fromJson(report.getStructuredOut(),
-              new TypeToken<Map<String, String>>() {
-              }.getType());
+          writeKeytabsStructuredOut = gson.fromJson(report.getStructuredOut(), WriteKeytabsStructuredOut.class);
         } catch (JsonSyntaxException ex) {
           //Json structure was incorrect do nothing, pass this data further for processing
-          structuredOutput = null;
+          writeKeytabsStructuredOut = null;
         }
 
-        if (structuredOutput != null) {
-          for (Map.Entry<String, String> entry : structuredOutput.entrySet()) {
-            String principal = entry.getKey();
-            if (!kerberosPrincipalHostDAO.exists(principal, hostname)) {
-              kerberosPrincipalHostDAO.create(principal, hostname);
+        if (writeKeytabsStructuredOut != null) {
+          Map<String, String> keytabs = writeKeytabsStructuredOut.getKeytabs();
+          if (keytabs != null) {
+            for (Map.Entry<String, String> entry : keytabs.entrySet()) {
+              String principal = entry.getKey();
+              if (!kerberosPrincipalHostDAO.exists(principal, hostname)) {
+                kerberosPrincipalHostDAO.create(principal, hostname);
+              }
             }
           }
         }
@@ -1102,4 +1103,20 @@ public class HeartBeatHandler {
     }
   }
 
+  /**
+   * This class is used for mapping json of structured output for keytab distribution actions.
+   */
+  private static class WriteKeytabsStructuredOut {
+    @SerializedName("keytabs")
+    private Map<String,String> keytabs;
+
+    public Map<String, String> getKeytabs() {
+      return keytabs;
+    }
+
+    public void setKeytabs(Map<String, String> keytabs) {
+      this.keytabs = keytabs;
+    }
+  }
+
 }

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

@@ -65,6 +65,7 @@ import org.apache.ambari.server.security.authorization.Users;
 import org.apache.ambari.server.security.ldap.AmbariLdapDataPopulator;
 import org.apache.ambari.server.security.ldap.LdapBatchDto;
 import org.apache.ambari.server.security.ldap.LdapSyncDto;
+import org.apache.ambari.server.serveraction.kerberos.KerberosInvalidConfigurationException;
 import org.apache.ambari.server.stageplanner.RoleGraph;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
@@ -1834,7 +1835,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     }
   }
 
-  private List<Stage> doStageCreation(RequestStageContainer requestStages,
+  private RequestStageContainer doStageCreation(RequestStageContainer requestStages,
       Cluster cluster,
       Map<State, List<Service>> changedServices,
       Map<State, List<ServiceComponent>> changedComps,
@@ -1856,7 +1857,8 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     if ((changedServices == null || changedServices.isEmpty())
         && (changedComps == null || changedComps.isEmpty())
         && (changedScHosts == null || changedScHosts.isEmpty())) {
-      return null;
+      LOG.debug("Created 0 stages");
+      return requestStages;
     }
 
     // smoke test any service that goes from installed to started
@@ -1882,10 +1884,12 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
       String HostParamsJson = StageUtils.getGson().toJson(
           customCommandExecutionHelper.createDefaultHostParams(cluster));
 
-      Stage stage = createNewStage(requestStages.getLastStageId() + 1, cluster,
+      Stage stage = createNewStage(requestStages.getLastStageId(), cluster,
           requestStages.getId(), requestProperties.get(REQUEST_CONTEXT_PROPERTY),
           clusterHostInfoJson, "{}", HostParamsJson);
 
+      Collection<ServiceComponentHost> componentsToEnableKerberos = new ArrayList<ServiceComponentHost>();
+
       //HACK
       String jobtrackerHost = getJobTrackerHost(cluster);
       for (String compName : changedScHosts.keySet()) {
@@ -1923,6 +1927,22 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
                       scHost.getServiceComponentName(), scHost.getHostName(),
                       nowTimestamp,
                       scHost.getDesiredStackVersion().getStackId());
+
+                  // If the state is transitioning from INIT TO INSTALLED and the cluster has Kerberos
+                  // enabled, mark this ServiceComponentHost to see if anything needs to be done to
+                  // make sure it is properly configured.  The Kerberos-related stages needs to be
+                  // between the INSTALLED and STARTED states because some services need to set up
+                  // the host (i,e, create user accounts, etc...) before Kerberos-related tasks an
+                  // occur (like distribute keytabs)
+                  if((oldSchState == State.INIT) && kerberosHelper.isClusterKerberosEnabled(cluster)) {
+                    try {
+                      kerberosHelper.configureService(cluster, scHost);
+                    } catch (KerberosInvalidConfigurationException e) {
+                      throw new AmbariException(e.getMessage(), e);
+                    }
+
+                    componentsToEnableKerberos.add(scHost);
+                  }
                 } else if (oldSchState == State.STARTED
                       // TODO: oldSchState == State.INSTALLED is always false, looks like a bug
                       //|| oldSchState == State.INSTALLED
@@ -2102,14 +2122,38 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
 
       RoleCommandOrder rco = getRoleCommandOrder(cluster);
       RoleGraph rg = new RoleGraph(rco);
+
       rg.build(stage);
-      return rg.getStages();
+      requestStages.addStages(rg.getStages());
+
+      if (!componentsToEnableKerberos.isEmpty()) {
+        Map<String, Collection<String>> serviceFilter = new HashMap<String, Collection<String>>();
+
+        for (ServiceComponentHost scHost : componentsToEnableKerberos) {
+          String serviceName = scHost.getServiceName();
+          Collection<String> componentFilter = serviceFilter.get(serviceName);
+
+          if (componentFilter == null) {
+            componentFilter = new HashSet<String>();
+            serviceFilter.put(serviceName, componentFilter);
+          }
+
+          componentFilter.add(scHost.getServiceComponentName());
+        }
+
+        kerberosHelper.ensureIdentities(cluster, serviceFilter, null, requestStages);
+      }
+
+      List<Stage> stages = requestStages.getStages();
+      LOG.debug("Created {} stages", ((stages != null) ? stages.size() : 0));
+
+    } else {
+      LOG.debug("Created 0 stages");
     }
 
-    return null;
+    return requestStages;
   }
 
-
   @Transactional
   void updateServiceStates(
       Map<State, List<Service>> changedServices,
@@ -2186,12 +2230,10 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
       requestStages = new RequestStageContainer(actionManager.getNextRequestId(), null, requestFactory, actionManager);
     }
 
-    List<Stage> stages = doStageCreation(requestStages, cluster, changedServices, changedComponents,
+    requestStages = doStageCreation(requestStages, cluster, changedServices, changedComponents,
         changedHosts, requestParameters, requestProperties,
         runSmokeTest, reconfigureClients);
-    LOG.debug("Created {} stages", ((stages != null) ? stages.size() : 0));
 
-    requestStages.addStages(stages);
     updateServiceStates(changedServices, changedComponents, changedHosts, ignoredHosts);
     return requestStages;
   }

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

@@ -198,6 +198,23 @@ public class AuthToLocalBuilder {
     return new Rule(rule.startsWith("RULE:") ? rule : String.format("RULE:%s", rule));
   }
 
+  /**
+   * Creates and returns a deep copy of this AuthToLocalBuilder.
+   *
+   * @return a deep copy of this AuthToLocalBuilder
+   */
+  public AuthToLocalBuilder copy() {
+    AuthToLocalBuilder copy = new AuthToLocalBuilder();
+
+    // TODO: This needs to be done in a loop rather than use Set.addAll because there may be an issue
+    // TODO: with the Rule.compareTo method?
+    for(Rule rule:setRules) {
+      copy.setRules.add(rule);
+    }
+
+    return copy;
+  }
+
 
   /**
    * Rule implementation.

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

@@ -114,7 +114,7 @@ public class KerberosHelper {
   private static final Logger LOG = LoggerFactory.getLogger(KerberosHelper.class);
 
   /**
-   * config type which contains the property used to determine if keberos is enabled
+   * config type which contains the property used to determine if Kerberos is enabled
    */
   private static final String SECURITY_ENABLED_CONFIG_TYPE = "cluster-env";
 
@@ -297,7 +297,7 @@ public class KerberosHelper {
    * executed to complete this task; or null if no stages need to be executed.
    * @throws AmbariException
    */
-  public RequestStageContainer ensureIdentities(Cluster cluster, Map<String, Collection<String>> serviceComponentFilter,
+  public RequestStageContainer ensureIdentities(Cluster cluster, Map<String, ? extends Collection<String>> serviceComponentFilter,
                                                 Collection<String> identityFilter, RequestStageContainer requestStageContainer)
       throws AmbariException {
     try {
@@ -308,6 +308,161 @@ public class KerberosHelper {
     }
   }
 
+  /**
+   * Updates the relevant configurations for the given Service.
+   * <p/>
+   * If the relevant service and its components have Kerberos descriptors, configuration values from
+   * the descriptors are used to update the relevant configuration sets.
+   *
+   * @param cluster              the relevant Cluster
+   * @param serviceComponentHost the ServiceComponentHost
+   * @throws AmbariException
+   */
+  public void configureService(Cluster cluster, ServiceComponentHost serviceComponentHost)
+      throws AmbariException, KerberosInvalidConfigurationException {
+
+    KerberosDetails kerberosDetails = getKerberosDetails(cluster);
+
+    // Set properties...
+    String serviceName = serviceComponentHost.getServiceName();
+    KerberosDescriptor kerberosDescriptor = getKerberosDescriptor(cluster);
+    KerberosServiceDescriptor serviceDescriptor = kerberosDescriptor.getService(serviceName);
+
+    if (serviceDescriptor != null) {
+      Map<String, String> kerberosDescriptorProperties = kerberosDescriptor.getProperties();
+      Map<String, Map<String, String>> kerberosConfigurations = new HashMap<String, Map<String, String>>();
+      Map<String, Map<String, String>> configurations = calculateConfigurations(cluster,
+          serviceComponentHost.getHostName(), kerberosDescriptorProperties);
+
+      Map<String, KerberosComponentDescriptor> componentDescriptors = serviceDescriptor.getComponents();
+      for (KerberosComponentDescriptor componentDescriptor : componentDescriptors.values()) {
+        if (componentDescriptor != null) {
+          Map<String, Map<String, String>> identityConfigurations;
+          List<KerberosIdentityDescriptor> identities;
+
+          identities = serviceDescriptor.getIdentities(true);
+          identityConfigurations = getConfigurations(identities);
+          if (identityConfigurations != null) {
+            for (Map.Entry<String, Map<String, String>> entry : identityConfigurations.entrySet()) {
+              mergeConfigurations(kerberosConfigurations, entry.getKey(), entry.getValue(), configurations);
+            }
+          }
+
+          identities = componentDescriptor.getIdentities(true);
+          identityConfigurations = getConfigurations(identities);
+          if (identityConfigurations != null) {
+            for (Map.Entry<String, Map<String, String>> entry : identityConfigurations.entrySet()) {
+              mergeConfigurations(kerberosConfigurations, entry.getKey(), entry.getValue(), configurations);
+            }
+          }
+
+          mergeConfigurations(kerberosConfigurations,
+              componentDescriptor.getConfigurations(true), configurations);
+        }
+      }
+
+      setAuthToLocalRules(kerberosDescriptor, cluster, kerberosDetails.getDefaultRealm(), configurations, kerberosConfigurations);
+
+      for (Map.Entry<String, Map<String, String>> entry : kerberosConfigurations.entrySet()) {
+        configHelper.updateConfigType(cluster, ambariManagementController, entry.getKey(), entry.getValue(),
+            ambariManagementController.getAuthName(), String.format("Enabling Kerberos for %s", serviceName));
+      }
+    }
+  }
+
+  /**
+   * Sets the relevant auth-to-local rule configuration properties using the services installed on
+   * the cluster and their relevant Kerberos descriptors to determine the rules to be created.
+   *
+   * @param kerberosDescriptor     the current Kerberos descriptor
+   * @param cluster                the cluster
+   * @param realm                  the default realm
+   * @param existingConfigurations a map of the current configurations
+   * @param kerberosConfigurations a map of the configurations to update, this where the generated
+   *                               auth-to-local values will be stored
+   * @throws AmbariException
+   */
+  private void setAuthToLocalRules(KerberosDescriptor kerberosDescriptor, Cluster cluster, String realm,
+                                   Map<String, Map<String, String>> existingConfigurations,
+                                   Map<String, Map<String, String>> kerberosConfigurations)
+      throws AmbariException {
+
+    if (kerberosDescriptor != null) {
+
+      Set<String> authToLocalProperties;
+      Set<String> authToLocalPropertiesToSet = new HashSet<String>();
+
+      // Determine which properties need to be set
+      AuthToLocalBuilder authToLocalBuilder = new AuthToLocalBuilder();
+
+      addIdentities(authToLocalBuilder, kerberosDescriptor.getIdentities(), null, existingConfigurations);
+
+      authToLocalProperties = kerberosDescriptor.getAuthToLocalProperties();
+      if (authToLocalProperties != null) {
+        authToLocalPropertiesToSet.addAll(authToLocalProperties);
+      }
+
+      Map<String, KerberosServiceDescriptor> services = kerberosDescriptor.getServices();
+      if (services != null) {
+        Map<String, Service> installedServices = cluster.getServices();
+
+        for (KerberosServiceDescriptor service : services.values()) {
+          if (installedServices.containsKey(service.getName())) {
+
+            addIdentities(authToLocalBuilder, service.getIdentities(true), null, existingConfigurations);
+
+            authToLocalProperties = service.getAuthToLocalProperties();
+            if (authToLocalProperties != null) {
+              authToLocalPropertiesToSet.addAll(authToLocalProperties);
+            }
+
+            Map<String, KerberosComponentDescriptor> components = service.getComponents();
+            if (components != null) {
+              for (KerberosComponentDescriptor component : components.values()) {
+                addIdentities(authToLocalBuilder, component.getIdentities(true), null, existingConfigurations);
+
+                authToLocalProperties = component.getAuthToLocalProperties();
+                if (authToLocalProperties != null) {
+                  authToLocalPropertiesToSet.addAll(authToLocalProperties);
+                }
+              }
+            }
+          }
+        }
+      }
+
+      if (!authToLocalPropertiesToSet.isEmpty()) {
+        for (String authToLocalProperty : authToLocalPropertiesToSet) {
+          String[] parts = authToLocalProperty.split("/");
+
+          if (parts.length == 2) {
+            AuthToLocalBuilder builder = authToLocalBuilder.copy();
+            String configType = parts[0];
+            String propertyName = parts[1];
+
+            // Add existing auth_to_local configuration, if set
+            Map<String, String> existingConfiguration = existingConfigurations.get(configType);
+            if (existingConfiguration != null) {
+              builder.addRules(existingConfiguration.get(propertyName));
+            }
+
+            // Add/update descriptor auth_to_local configuration, if set
+            Map<String, String> kerberosConfiguration = kerberosConfigurations.get(configType);
+            if (kerberosConfiguration != null) {
+              builder.addRules(kerberosConfiguration.get(propertyName));
+            } else {
+              kerberosConfiguration = new HashMap<String, String>();
+              kerberosConfigurations.put(configType, kerberosConfiguration);
+            }
+
+            kerberosConfiguration.put(propertyName, builder.generate(realm));
+          }
+        }
+      }
+    }
+  }
+
+
   /**
    * Performs operations needed to process Kerberos related tasks on the relevant cluster.
    * <p/>
@@ -334,7 +489,7 @@ public class KerberosHelper {
   @Transactional
   private RequestStageContainer handle(Cluster cluster,
                                        KerberosDetails kerberosDetails,
-                                       Map<String, Collection<String>> serviceComponentFilter,
+                                       Map<String, ? extends Collection<String>> serviceComponentFilter,
                                        Collection<String> identityFilter,
                                        RequestStageContainer requestStageContainer,
                                        Handler handler) throws AmbariException {
@@ -353,7 +508,6 @@ public class KerberosHelper {
         KerberosActionDataFileBuilder kerberosActionDataFileBuilder = null;
         Map<String, String> kerberosDescriptorProperties = kerberosDescriptor.getProperties();
         Map<String, Map<String, String>> kerberosConfigurations = new HashMap<String, Map<String, String>>();
-        AuthToLocalBuilder authToLocalBuilder = new AuthToLocalBuilder();
 
         // Create a temporary directory to store metadata needed to complete this task.  Information
         // such as which principals and keytabs files to create as well as what configurations need
@@ -384,38 +538,7 @@ public class KerberosHelper {
             if ((serviceComponentHosts != null) && !serviceComponentHosts.isEmpty()) {
               // Calculate the current host-specific configurations. These will be used to replace
               // variables within the Kerberos descriptor data
-              Map<String, Map<String, String>> configurations = calculateConfigurations(cluster, hostname);
-              // add existing kerberos auth_to_local rules to builder
-              if (configurations.containsKey("core-site")) {
-                authToLocalBuilder.addRules(
-                    configurations.get("core-site").get("hadoop.security.auth_to_local"));
-              }
-
-              // A map to hold un-categorized properties.  This may come from the KerberosDescriptor
-              // and will also contain a value for the current host
-              Map<String, String> generalProperties = new HashMap<String, String>();
-
-              // Make sure the configurations exist.
-              if (configurations == null) {
-                configurations = new HashMap<String, Map<String, String>>();
-              }
-
-              // If any properties are set in the calculated KerberosDescriptor, add them into the
-              // Map of configurations as an un-categorized type (using an empty string)
-              if (kerberosDescriptorProperties != null) {
-                generalProperties.putAll(kerberosDescriptorProperties);
-              }
-
-              // Add the current hostname under "host" and "hostname"
-              generalProperties.put("host", hostname);
-              generalProperties.put("hostname", hostname);
-              generalProperties.put("cluster_name", clusterName);
-
-              if (configurations.get("") == null) {
-                configurations.put("", generalProperties);
-              } else {
-                configurations.get("").putAll(generalProperties);
-              }
+              Map<String, Map<String, String>> configurations = calculateConfigurations(cluster, hostname, kerberosDescriptorProperties);
 
               // Iterate over the components installed on the current host to get the service and
               // component-level Kerberos descriptors in order to determine which principals,
@@ -430,45 +553,39 @@ public class KerberosHelper {
 
                   if (serviceDescriptor != null) {
                     String componentName = sch.getServiceComponentName();
+                    int identitiesAdded = 0;
+                    List<KerberosIdentityDescriptor> serviceIdentities = serviceDescriptor.getIdentities(true);
+
+                    // Lazily create the KerberosActionDataFileBuilder instance...
+                    if (kerberosActionDataFileBuilder == null) {
+                      kerberosActionDataFileBuilder = new KerberosActionDataFileBuilder(indexFile);
+                    }
+
+                    // Add service-level principals (and keytabs)
+                    identitiesAdded += addIdentities(kerberosActionDataFileBuilder, serviceIdentities,
+                        identityFilter, hostname, serviceName, componentName, configurations);
 
                     // If there is no filter or the filter contains the current component name,
                     // test to see if this component should be process by querying the handler...
                     if (((componentFilter == null) || componentFilter.contains(componentName)) && handler.shouldProcess(desiredSecurityState, sch)) {
                       KerberosComponentDescriptor componentDescriptor = serviceDescriptor.getComponent(componentName);
-                      List<KerberosIdentityDescriptor> serviceIdentities = serviceDescriptor.getIdentities(true);
 
                       if (componentDescriptor != null) {
                         List<KerberosIdentityDescriptor> componentIdentities = componentDescriptor.getIdentities(true);
-                        int identitiesAdded = 0;
 
                         // Calculate the set of configurations to update and replace any variables
                         // using the previously calculated Map of configurations for the host.
                         mergeConfigurations(kerberosConfigurations,
                             componentDescriptor.getConfigurations(true), configurations);
 
-                        // Lazily create the KerberosActionDataFileBuilder instance...
-                        if (kerberosActionDataFileBuilder == null) {
-                          kerberosActionDataFileBuilder = new KerberosActionDataFileBuilder(indexFile);
-                        }
-
-                        // Add service-level principals (and keytabs)
-                        identitiesAdded += addIdentities(kerberosActionDataFileBuilder, serviceIdentities,
-                            identityFilter, hostname, serviceName, componentName, configurations);
-
                         // Add component-level principals (and keytabs)
                         identitiesAdded += addIdentities(kerberosActionDataFileBuilder, componentIdentities,
                             identityFilter, hostname, serviceName, componentName, configurations);
-
-                        if (identitiesAdded > 0) {
-                          serviceComponentHostsToProcess.add(sch);
-                        }
-
-                        // Add component-level principals to auth_to_local builder
-                        addIdentities(authToLocalBuilder, componentIdentities, identityFilter, configurations);
                       }
+                    }
 
-                      // Add service-level principals to auth_to_local builder
-                      addIdentities(authToLocalBuilder, serviceIdentities, identityFilter, configurations);
+                    if (identitiesAdded > 0) {
+                      serviceComponentHostsToProcess.add(sch);
                     }
                   }
                 }
@@ -506,20 +623,9 @@ public class KerberosHelper {
             throw new IllegalArgumentException(e.getMessage(), e);
           }
 
-          // Determine if the any auth_to_local configurations need to be set dynamically
-          // Lazily create the auth_to_local rules
-          String authToLocal = null;
-          for (Map<String, String> configuration : kerberosConfigurations.values()) {
-            for (Map.Entry<String, String> entry : configuration.entrySet()) {
-              if ("_AUTH_TO_LOCAL_RULES".equals(entry.getValue())) {
-                if (authToLocal == null) {
-                  authToLocal = authToLocalBuilder.generate(kerberosDetails.getDefaultRealm());
-                }
-
-                entry.setValue(authToLocal);
-              }
-            }
-          }
+          setAuthToLocalRules(kerberosDescriptor, cluster, kerberosDetails.getDefaultRealm(),
+              calculateConfigurations(cluster, null, kerberosDescriptorProperties),
+              kerberosConfigurations);
         }
 
         // Ensure the cluster-env/security_enabled flag is set properly
@@ -561,6 +667,7 @@ public class KerberosHelper {
 
         // Add the cleanup stage...
         Map<String, String> commandParameters = new HashMap<String, String>();
+        commandParameters.put(KerberosServerAction.AUTHENTICATED_USER_NAME, ambariManagementController.getAuthName());
         commandParameters.put(KerberosServerAction.DATA_DIRECTORY, dataDirectory.getAbsolutePath());
 
         Stage stage = createServerActionStage(requestStageContainer.getLastStageId(),
@@ -944,21 +1051,7 @@ public class KerberosHelper {
 
         if (configurationDescriptor != null) {
           Map<String, String> updatedProperties = configurationDescriptor.getProperties();
-
-          if (updatedProperties != null) {
-            Map<String, String> existingProperties = configurations.get(type);
-            if (existingProperties == null) {
-              existingProperties = new HashMap<String, String>();
-              configurations.put(type, existingProperties);
-            }
-
-            for (Map.Entry<String, String> property : updatedProperties.entrySet()) {
-              existingProperties.put(
-                  KerberosDescriptor.replaceVariables(property.getKey(), replacements),
-                  KerberosDescriptor.replaceVariables(property.getValue(), replacements)
-              );
-            }
-          }
+          mergeConfigurations(configurations, type, updatedProperties, replacements);
         }
       }
     }
@@ -966,6 +1059,25 @@ public class KerberosHelper {
     return configurations;
   }
 
+  private void mergeConfigurations(Map<String, Map<String, String>> configurations, String type,
+                                   Map<String, String> updates,
+                                   Map<String, Map<String, String>> replacements) throws AmbariException {
+    if (updates != null) {
+      Map<String, String> existingProperties = configurations.get(type);
+      if (existingProperties == null) {
+        existingProperties = new HashMap<String, String>();
+        configurations.put(type, existingProperties);
+      }
+
+      for (Map.Entry<String, String> property : updates.entrySet()) {
+        existingProperties.put(
+            KerberosDescriptor.replaceVariables(property.getKey(), replacements),
+            KerberosDescriptor.replaceVariables(property.getValue(), replacements)
+        );
+      }
+    }
+  }
+
   /**
    * Adds identities to the KerberosActionDataFileBuilder.
    *
@@ -1081,22 +1193,23 @@ public class KerberosHelper {
   /**
    * Calculates the map of configurations relative to the cluster and host.
    * <p/>
-   * This was borrowed from {@link org.apache.ambari.server.actionmanager.ExecutionCommandWrapper#getExecutionCommand()}
+   * Most of this was borrowed from {@link org.apache.ambari.server.actionmanager.ExecutionCommandWrapper#getExecutionCommand()}
    *
-   * @param cluster  the relevant Cluster
-   * @param hostname the relevant hostname
+   * @param cluster                      the relevant Cluster
+   * @param hostname                     the relevant hostname
+   * @param kerberosDescriptorProperties a map of general Kerberos descriptor properties
    * @return a Map of calculated configuration types
    * @throws AmbariException
    */
-  private Map<String, Map<String, String>> calculateConfigurations(Cluster cluster, String hostname) throws AmbariException {
+  private Map<String, Map<String, String>> calculateConfigurations(Cluster cluster, String hostname,
+                                                                   Map<String, String> kerberosDescriptorProperties)
+      throws AmbariException {
     // For a configuration type, both tag and an actual configuration can be stored
     // Configurations from the tag is always expanded and then over-written by the actual
     // global:version1:{a1:A1,b1:B1,d1:D1} + global:{a1:A2,c1:C1,DELETED_d1:x} ==>
     // global:{a1:A2,b1:B1,c1:C1}
     Map<String, Map<String, String>> configurations = new HashMap<String, Map<String, String>>();
-
     Map<String, Map<String, String>> configurationTags = ambariManagementController.findConfigurationTagsWithOverrides(cluster, hostname);
-    Map<String, Map<String, Map<String, String>>> configurationAttributes = new TreeMap<String, Map<String, Map<String, String>>>();
 
     if (configurationTags.get(Configuration.GLOBAL_CONFIG_TAG) != null) {
       configHelper.applyCustomConfig(
@@ -1124,26 +1237,34 @@ public class KerberosHelper {
       configurations.put(type, configuration);
     }
 
-    Map<String, Map<String, Map<String, String>>> configAttributes =
-        configHelper.getEffectiveConfigAttributes(cluster, configurationTags);
-
-    for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) {
-      String type = attributesOccurrence.getKey();
-      Map<String, Map<String, String>> attributes = attributesOccurrence.getValue();
+    // A map to hold un-categorized properties.  This may come from the KerberosDescriptor
+    // and will also contain a value for the current host
+    Map<String, String> generalProperties = configurations.get("");
+    if (generalProperties == null) {
+      generalProperties = new HashMap<String, String>();
+      configurations.put("", generalProperties);
+    }
 
-      if (!configurationAttributes.containsKey(type)) {
-        configurationAttributes.put(type, new TreeMap<String, Map<String, String>>());
-      }
-      configHelper.cloneAttributesMap(attributes, configurationAttributes.get(type));
+    // If any properties are set in the calculated KerberosDescriptor, add them into the
+    // Map of configurations as an un-categorized type (using an empty string)
+    if (kerberosDescriptorProperties != null) {
+      generalProperties.putAll(kerberosDescriptorProperties);
     }
 
+    // Add the current hostname under "host" and "hostname"
+    generalProperties.put("host", hostname);
+    generalProperties.put("hostname", hostname);
+
+    // Add the current cluster's name
+    generalProperties.put("cluster_name", cluster.getClusterName());
+
     // add clusterHostInfo config
     Map<String, String> componentHosts = new HashMap<String, String>();
     for (Map.Entry<String, Service> service : cluster.getServices().entrySet()) {
       for (Map.Entry<String, ServiceComponent> serviceComponent : service.getValue().getServiceComponents().entrySet()) {
         if (StageUtils.getComponentToClusterInfoKeyMap().keySet().contains(serviceComponent.getValue().getName())) {
           componentHosts.put(StageUtils.getComponentToClusterInfoKeyMap().get(serviceComponent.getValue().getName()),
-                  StringUtils.join(serviceComponent.getValue().getServiceComponentHosts().keySet(), ","));
+              StringUtils.join(serviceComponent.getValue().getServiceComponentHosts().keySet(), ","));
         }
       }
     }
@@ -1302,6 +1423,62 @@ public class KerberosHelper {
     return cluster.getSecurityType() == SecurityType.KERBEROS;
   }
 
+  /**
+   * Given a list of KerberosIdentityDescriptors, returns a Map fo configuration types to property
+   * names and values.
+   * <p/>
+   * The property names and values are not expected to have any variable replacements done.
+   *
+   * @param identityDescriptors a List of KerberosIdentityDescriptor from which to retrieve configurations
+   * @return a Map of configuration types to property name/value pairs (as a Map)
+   */
+  private Map<String, Map<String, String>> getConfigurations(List<KerberosIdentityDescriptor> identityDescriptors) {
+    Map<String, Map<String, String>> map = new HashMap<String, Map<String, String>>();
+
+    if (identityDescriptors != null) {
+      for (KerberosIdentityDescriptor identityDescriptor : identityDescriptors) {
+        KerberosPrincipalDescriptor principalDescriptor = identityDescriptor.getPrincipalDescriptor();
+        if (principalDescriptor != null) {
+          putConfiguration(map, principalDescriptor.getConfiguration(), principalDescriptor.getValue());
+        }
+
+        KerberosKeytabDescriptor keytabDescriptor = identityDescriptor.getKeytabDescriptor();
+        if (keytabDescriptor != null) {
+          putConfiguration(map, keytabDescriptor.getConfiguration(), keytabDescriptor.getFile());
+        }
+      }
+    }
+
+    return map;
+  }
+
+  /**
+   * Inserts a configuration property and value into a map of configuration types to property
+   * name/value pair maps.
+   *
+   * @param map           the Map to insert into
+   * @param configuration a configuration property in the form of config-type/property_name
+   * @param value         the value of the configuration property
+   */
+  private void putConfiguration(Map<String, Map<String, String>> map, String configuration, String value) {
+    if (configuration != null) {
+      String[] principalTokens = configuration.split("/");
+
+      if (principalTokens.length == 2) {
+        String type = principalTokens[0];
+        String propertyName = principalTokens[1];
+
+        Map<String, String> properties = map.get(type);
+        if (properties == null) {
+          properties = new HashMap<String, String>();
+          map.put(type, properties);
+        }
+
+        properties.put(propertyName, value);
+      }
+    }
+  }
+
   /**
    * A enumeration of the supported custom operations
    */
@@ -1612,6 +1789,7 @@ public class KerberosHelper {
       }
 
       Map<String, String> commandParameters = new HashMap<String, String>();
+      commandParameters.put(KerberosServerAction.AUTHENTICATED_USER_NAME, ambariManagementController.getAuthName());
       commandParameters.put(KerberosServerAction.DATA_DIRECTORY, dataDirectory.getAbsolutePath());
       commandParameters.put(KerberosServerAction.DEFAULT_REALM, kerberosDetails.getDefaultRealm());
       commandParameters.put(KerberosServerAction.KDC_TYPE, kerberosDetails.getKdcType().name());
@@ -1701,6 +1879,7 @@ public class KerberosHelper {
       }
 
       Map<String, String> commandParameters = new HashMap<String, String>();
+      commandParameters.put(KerberosServerAction.AUTHENTICATED_USER_NAME, ambariManagementController.getAuthName());
       commandParameters.put(KerberosServerAction.DATA_DIRECTORY, dataDirectory.getAbsolutePath());
       commandParameters.put(KerberosServerAction.DEFAULT_REALM, kerberosDetails.getDefaultRealm());
       commandParameters.put(KerberosServerAction.KDC_TYPE, kerberosDetails.getKdcType().name());
@@ -1863,6 +2042,7 @@ public class KerberosHelper {
       }
 
       Map<String, String> commandParameters = new HashMap<String, String>();
+      commandParameters.put(KerberosServerAction.AUTHENTICATED_USER_NAME, ambariManagementController.getAuthName());
       commandParameters.put(KerberosServerAction.DATA_DIRECTORY, dataDirectory.getAbsolutePath());
       commandParameters.put(KerberosServerAction.DEFAULT_REALM, kerberosDetails.getDefaultRealm());
       commandParameters.put(KerberosServerAction.KDC_TYPE, kerberosDetails.getKdcType().name());

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

@@ -34,7 +34,6 @@ import com.google.inject.Injector;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
 import org.apache.ambari.server.controller.AmbariManagementController;
-import org.apache.ambari.server.controller.KerberosHelper;
 import org.apache.ambari.server.controller.MaintenanceStateHelper;
 import org.apache.ambari.server.controller.RequestStatusResponse;
 import org.apache.ambari.server.controller.ServiceComponentHostRequest;
@@ -60,7 +59,6 @@ import com.google.inject.assistedinject.AssistedInject;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.MaintenanceState;
-import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.ServiceComponent;
 import org.apache.ambari.server.state.ServiceComponentHost;
 import org.apache.ambari.server.state.ServiceComponentHostEvent;
@@ -124,13 +122,6 @@ public class HostComponentResourceProvider extends AbstractControllerResourcePro
   @Inject
   private MaintenanceStateHelper maintenanceStateHelper;
 
-  /**
-   * kerberos helper
-   */
-  @Inject
-  private KerberosHelper kerberosHelper;
-
-
   // ----- Constructors ----------------------------------------------------
 
   /**
@@ -415,8 +406,6 @@ public class HostComponentResourceProvider extends AbstractControllerResourcePro
       clusterNames.add(clusterName);
     }
 
-    boolean addKerberosStages = false;
-
     for (ServiceComponentHostRequest request : requests) {
       validateServiceComponentHostRequest(request);
 
@@ -475,10 +464,6 @@ public class HostComponentResourceProvider extends AbstractControllerResourcePro
         // set desired state on host component
         newState = State.valueOf(request.getDesiredState());
 
-        // determine if this state transition will require that kerberos stages are added to request.
-        // once set to true will stay true
-        addKerberosStages = addKerberosStages || requiresKerberosStageAddition(oldState, newState, cluster);
-
         // throw exception if desired state isn't a valid desired state (static check)
         if (!newState.isValidDesiredState()) {
           throw new IllegalArgumentException("Invalid arguments, invalid"
@@ -563,16 +548,9 @@ public class HostComponentResourceProvider extends AbstractControllerResourcePro
     // just getting the first cluster
     Cluster cluster = clusters.getCluster(clusterNames.iterator().next());
 
-    RequestStageContainer requestStages = getManagementController().addStages(
+    return getManagementController().addStages(
         stages, cluster, requestProperties, null, null, null,
         changedScHosts, ignoredScHosts, runSmokeTest, false);
-
-    if (addKerberosStages) {
-      // adds the necessary kerberos related stages to the request
-      kerberosHelper.toggleKerberos(cluster, SecurityType.KERBEROS, requestStages);
-    }
-
-    return requestStages;
   }
 
   @Override
@@ -847,21 +825,6 @@ public class HostComponentResourceProvider extends AbstractControllerResourcePro
     }
   }
 
-  /**
-   * Determine if kerberos stages need to be added to the request as a result of a
-   * host component state change.
-   *
-   * @param current  current host component state
-   * @param target   target host component state
-   * @param cluster  associated cluster
-   * @return whether kerberos stages should be added to the request
-   */
-  public boolean requiresKerberosStageAddition(State current, State target, Cluster cluster) {
-    return current == State.INIT &&
-        target  == State.INSTALLED &&
-        kerberosHelper.isClusterKerberosEnabled(cluster);
-  }
-
 
   // ----- inner classes ---------------------------------------------------
 

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/KerberosPrincipalHostDAO.java

@@ -165,7 +165,7 @@ public class KerberosPrincipalHostDAO {
    */
   @Transactional
   public void removeByPrincipal(String principalName) {
-    entityManagerProvider.get().remove(findByPrincipal(principalName));
+    remove(findByPrincipal(principalName));
   }
 
   /**
@@ -175,7 +175,7 @@ public class KerberosPrincipalHostDAO {
    */
   @Transactional
   public void removeByHost(String hostName) {
-    entityManagerProvider.get().remove(findByHost(hostName));
+    remove(findByHost(hostName));
   }
 
   /**
@@ -201,4 +201,18 @@ public class KerberosPrincipalHostDAO {
   public boolean exists(String principalName, String hostName) {
     return find(principalName, hostName) != null;
   }
+
+  /**
+   * Removes multiple KerberosPrincipalHostEntity items
+   *
+   * @param entities a collection of KerberosPrincipalHostEntity items to remove
+   */
+  private void remove(List<KerberosPrincipalHostEntity> entities) {
+    if (entities != null) {
+      for (KerberosPrincipalHostEntity entity : entities) {
+        entityManagerProvider.get().remove(entity);
+      }
+    }
+  }
+
 }

ambari-server/src/main/java/org/apache/ambari/server/serveraction/kerberos/KerberosServerAction.java

@@ -45,6 +45,12 @@ import java.util.Map;
  * (see {@link org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFile}).
  */
 public abstract class KerberosServerAction extends AbstractServerAction {
+  /**
+   * A (command parameter) property name used to hold the authenticated user's name for use in
+   * operations that record the acting user.
+   */
+  public static final String AUTHENTICATED_USER_NAME = "authenticated_user_name";
+
   /**
    * A (command parameter) property name used to hold the absolute path to the directory that is to
    * be used to store transient data while the request is being processed.  This is expected to be

ambari-server/src/main/java/org/apache/ambari/server/serveraction/kerberos/UpdateKerberosConfigsServerAction.java

@@ -18,24 +18,21 @@
 
 package org.apache.ambari.server.serveraction.kerberos;
 
-import com.google.common.collect.Maps;
 import com.google.inject.Inject;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
 import org.apache.ambari.server.agent.CommandReport;
 import org.apache.ambari.server.controller.AmbariManagementController;
-import org.apache.ambari.server.controller.ConfigurationRequest;
 import org.apache.ambari.server.serveraction.AbstractServerAction;
 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.ConfigHelper;
 import org.apache.ambari.server.state.SecurityType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
@@ -46,14 +43,14 @@ import java.util.concurrent.ConcurrentMap;
  */
 public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
 
-  private final static Logger LOG =
-    LoggerFactory.getLogger(UpdateKerberosConfigsServerAction.class);
-
-  private HashMap<String, Map<String, String>> configurations = new HashMap<String, Map<String, String>>();
+  private final static Logger LOG = LoggerFactory.getLogger(UpdateKerberosConfigsServerAction.class);
 
   @Inject
   private AmbariManagementController controller;
 
+  @Inject
+  private ConfigHelper configHelper;
+
   /**
    * Executes this ServerAction
    * <p/>
@@ -65,12 +62,11 @@ public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
    *                                 to a given request
    * @return a CommandReport declaring the status of the task
    * @throws org.apache.ambari.server.AmbariException
-   *
    * @throws InterruptedException
    */
   @Override
   public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext)
-    throws AmbariException, InterruptedException {
+      throws AmbariException, InterruptedException {
 
     CommandReport commandReport = null;
 
@@ -78,7 +74,9 @@ public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
     Clusters clusters = controller.getClusters();
     Cluster cluster = clusters.getCluster(clusterName);
 
+    String authenticatedUserName = getCommandParameterValue(getCommandParameters(), KerberosServerAction.AUTHENTICATED_USER_NAME);
     String dataDirectoryPath = getCommandParameterValue(getCommandParameters(), KerberosServerAction.DATA_DIRECTORY);
+    HashMap<String, Map<String, String>> configurations = new HashMap<String, Map<String, String>>();
 
     // If the data directory path is set, attempt to process further, else assume there is no work to do
     if (dataDirectoryPath != null) {
@@ -103,7 +101,7 @@ public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
               if (principalTokens.length == 2) {
                 String principalConfigType = principalTokens[0];
                 String principalConfigProp = principalTokens[1];
-                addConfigTypePropVal(principalConfigType, principalConfigProp, principal);
+                addConfigTypePropVal(configurations, principalConfigType, principalConfigProp, principal);
               }
 
               String keytabPath = record.get(KerberosActionDataFile.KEYTAB_FILE_PATH);
@@ -112,7 +110,7 @@ public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
               if (keytabTokens.length == 2) {
                 String keytabConfigType = keytabTokens[0];
                 String keytabConfigProp = keytabTokens[1];
-                addConfigTypePropVal(keytabConfigType, keytabConfigProp, keytabPath);
+                addConfigTypePropVal(configurations, keytabConfigType, keytabConfigProp, keytabPath);
               }
             }
           }
@@ -126,18 +124,19 @@ public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
               String configType = record.get(KerberosConfigDataFile.CONFIGURATION_TYPE);
               String configKey = record.get(KerberosConfigDataFile.KEY);
               String configVal = record.get(KerberosConfigDataFile.VALUE);
-              addConfigTypePropVal(configType, configKey, configVal);
+              addConfigTypePropVal(configurations, configType, configKey, configVal);
             }
           }
 
-          for (Map.Entry<String, Map<String, String>> entry : configurations.entrySet()) {
-            updateConfigurationPropertiesForCluster(
-                cluster,
-                entry.getKey(),     // configType
-                entry.getValue(),   // properties
-                true,               // updateIfExists
-                true,               // createNew
-                "update services configs to enable kerberos");
+          if (!configurations.isEmpty()) {
+            String configNote = cluster.getSecurityType() == SecurityType.KERBEROS
+                ? "Enabling Kerberos"
+                : "Disabling Kerberos";
+
+            for (Map.Entry<String, Map<String, String>> entry : configurations.entrySet()) {
+                configHelper.updateConfigType(cluster, controller, entry.getKey(), entry.getValue(),
+                  authenticatedUserName, configNote);
+            }
           }
         } catch (IOException e) {
           String message = "Could not update services configs to enable kerberos";
@@ -170,116 +169,12 @@ public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
   }
 
 
-  /**
-   *
-   * Updates service config properties of a cluster
-   * @param cluster   the cluster for which to update service configs
-   * @param configType   service config type to be updated
-   * @param properties    map of service config properties
-   * @param updateIfExists    flag indicating whether to update if a property already exists
-   * @param createNewConfigType  flag indicating whether to create new service config
-   *                             if the config type does not exist
-   * @param note   a short note on change
-   * @throws AmbariException if the operation fails
-   */
-  private void updateConfigurationPropertiesForCluster(
-    Cluster cluster,
-    String configType,
-    Map<String, String> properties,
-    boolean updateIfExists,
-    boolean createNewConfigType,
-    String note)
-    throws AmbariException {
-
-    String newTag = "version" + System.currentTimeMillis();
-    String message;
-    if ((properties != null) && (properties.size() > 0)) {
-      Map<String, Config> all = cluster.getConfigsByType(configType);
-      if (all == null || !all.containsKey(newTag)) {
-        Map<String, String> oldConfigProperties;
-        Config oldConfig = cluster.getDesiredConfigByType(configType);
-
-        if (oldConfig == null && !createNewConfigType) {
-          message = String.format("Config %s not found. Assuming service not installed. " +
-              "Skipping configuration properties update", configType);
-          actionLog.writeStdOut(message);
-          LOG.info(message);
-          return;
-        } else if (oldConfig == null) {
-          oldConfigProperties = new HashMap<String, String>();
-          newTag = "version1";
-        } else {
-          oldConfigProperties = oldConfig.getProperties();
-        }
-
-        Map<String, String> mergedProperties =
-          mergeProperties(oldConfigProperties, properties, updateIfExists);
-
-        if (!Maps.difference(oldConfigProperties, mergedProperties).areEqual()) {
-          message = String.format("Applying configuration with tag '%s' to " +
-              "cluster '%s'", newTag, cluster.getClusterName());
-          actionLog.writeStdOut(message);
-          LOG.info(message);
-
-          ConfigurationRequest cr = new ConfigurationRequest();
-          cr.setClusterName(cluster.getClusterName());
-          cr.setVersionTag(newTag);
-          cr.setType(configType);
-          cr.setProperties(mergedProperties);
-          cr.setServiceConfigVersionNote(note);
-          controller.createConfiguration(cr);
-
-          Config baseConfig = cluster.getConfig(cr.getType(), cr.getVersionTag());
-          if (baseConfig != null) {
-            String authName = controller.getAuthName();
-            String configNote = null;
-            configNote = cluster.getSecurityType() == SecurityType.KERBEROS ?
-                    "Enabling Kerberos on Cluster" : "Disabling Kerberos on Cluster";
-            if (cluster.addDesiredConfig(authName, Collections.singleton(baseConfig), configNote) != null) {
-              String oldConfigString = (oldConfig != null) ? " from='" + oldConfig.getTag() + "'" : "";
-              message = "cluster '" + cluster.getClusterName() + "' "
-                  + "changed by: '" + authName + "'; "
-                  + "type='" + baseConfig.getType() + "' "
-                  + "tag='" + baseConfig.getTag() + "'"
-                  + oldConfigString;
-              LOG.info(message);
-              actionLog.writeStdOut(message);
-            }
-          }
-        } else {
-          message = "No changes detected to config " + configType + ". Skipping configuration properties update";
-          LOG.info(message);
-          actionLog.writeStdOut(message);
-        }
-      }
-    }
-  }
-
-  /**
-   * Merges current properties and new properties
-   * @param originalProperties  current properties
-   * @param newProperties      new properties
-   * @param updateIfExists    flag indicating whether to update if a property already exists
-   * @return    merged properties
-   */
-  private static Map<String, String> mergeProperties(Map<String, String> originalProperties,
-                                                     Map<String, String> newProperties,
-                                                     boolean updateIfExists) {
-
-    Map<String, String> properties = new HashMap<String, String>(originalProperties);
-    for (Map.Entry<String, String> entry : newProperties.entrySet()) {
-      if (!properties.containsKey(entry.getKey()) || updateIfExists) {
-        properties.put(entry.getKey(), entry.getValue());
-      }
-    }
-    return properties;
-  }
-
   /**
    * Gets a property from the given commandParameters
-   * @param commandParameters   map of command parameters
-   * @param propertyName   property name to find value for
-   * @return    value of given proeprty name, would return <code>null</code>
+   *
+   * @param commandParameters map of command parameters
+   * @param propertyName      property name to find value for
+   * @return value of given proeprty name, would return <code>null</code>
    * if the provided commandParameters is null or  if the requested property is not found
    * in commandParams
    */
@@ -289,11 +184,13 @@ public class UpdateKerberosConfigsServerAction extends AbstractServerAction {
 
   /**
    * Adds a property to properties of a given service config type
-   * @param configtype   service config type
-   * @param prop     property to be added
-   * @param val   value for the proeprty
+   *
+   * @param configurations
+   * @param configtype     service config type
+   * @param prop           property to be added
+   * @param val            value for the proeprty
    */
-  private void addConfigTypePropVal(String configtype, String prop, String val) {
+  private void addConfigTypePropVal(HashMap<String, Map<String, String>> configurations, String configtype, String prop, String val) {
     Map<String, String> configtypePropsVal = configurations.get(configtype);
     if (configtypePropsVal == null) {
       configtypePropsVal = new HashMap<String, String>();

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

@@ -31,6 +31,7 @@ import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
 
+import com.google.common.collect.Maps;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
 import org.apache.ambari.server.configuration.Configuration;
@@ -638,6 +639,53 @@ public class ConfigHelper {
     return ambariMetaInfo.getStackProperties(stack.getName(), stack.getVersion());
   }
 
+  /**
+   * A helper method to create a new {@link Config} for a given configuration
+   * type and updates to the current values, if any. This method will perform the following tasks:
+   * <ul>
+   * <li>Marge the specified updates with the properties of the current version of the
+   * configuration</li>
+   * <li>Create a {@link Config} in the cluster for the specified type. This
+   * will have the proper versions and tags set automatically.</li>
+   * <li>Set the cluster's {@link DesiredConfig} to the new configuration</li>
+   * <li>Create an entry in the configuration history with a note and username.</li>
+   * <ul>
+   *
+   * @param cluster
+   * @param controller
+   * @param configType
+   * @param updates
+   * @param authenticatedUserName
+   * @param serviceVersionNote
+   * @throws AmbariException
+   */
+  public void updateConfigType(Cluster cluster,
+                               AmbariManagementController controller, String configType,
+                               Map<String, String> updates, String authenticatedUserName,
+                               String serviceVersionNote) throws AmbariException {
+
+    if((configType != null) && (updates != null) && !updates.isEmpty()) {
+      Config oldConfig = cluster.getDesiredConfigByType(configType);
+      Map<String, String> oldConfigProperties;
+      Map<String, String> properties = new HashMap<String, String>();
+
+      if (oldConfig == null) {
+        oldConfigProperties = null;
+      } else {
+        oldConfigProperties = oldConfig.getProperties();
+        if (oldConfigProperties != null) {
+          properties.putAll(oldConfig.getProperties());
+        }
+      }
+
+      properties.putAll(updates);
+
+      if ((oldConfigProperties == null) || !Maps.difference(oldConfigProperties, properties).areEqual()) {
+        createConfigType(cluster, controller, configType, properties, authenticatedUserName, serviceVersionNote);
+      }
+    }
+  }
+
   /**
    * A helper method to create a new {@link Config} for a given configuration
    * type. This method will perform the following tasks:

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

@@ -47,6 +47,7 @@ import org.apache.ambari.server.orm.dao.ClusterDAO;
 import org.apache.ambari.server.orm.dao.ClusterVersionDAO;
 import org.apache.ambari.server.orm.dao.HostDAO;
 import org.apache.ambari.server.orm.dao.HostVersionDAO;
+import org.apache.ambari.server.orm.dao.KerberosPrincipalHostDAO;
 import org.apache.ambari.server.orm.dao.ResourceTypeDAO;
 import org.apache.ambari.server.orm.entities.ClusterEntity;
 import org.apache.ambari.server.orm.entities.ClusterVersionEntity;
@@ -109,6 +110,8 @@ public class ClustersImpl implements Clusters {
   @Inject
   ResourceTypeDAO resourceTypeDAO;
   @Inject
+  KerberosPrincipalHostDAO kerberosPrincipalHostDAO;
+  @Inject
   ClusterFactory clusterFactory;
   @Inject
   HostFactory hostFactory;
@@ -692,6 +695,8 @@ public class ClustersImpl implements Clusters {
 
       deleteConfigGroupHostMapping(hostname);
 
+      // Remove mapping of principals to the unmapped host
+      kerberosPrincipalHostDAO.removeByHost(hostname);
     } finally {
       w.unlock();
     }
@@ -743,6 +748,9 @@ public class ClustersImpl implements Clusters {
       hostDAO.remove(entity);
       hosts.remove(hostname);
 
+      // Remove mapping of principals to deleted host
+      kerberosPrincipalHostDAO.removeByHost(hostname);
+
       // publish the event
       HostRemovedEvent event = new HostRemovedEvent(hostname);
       eventPublisher.publish(event);

ambari-server/src/main/java/org/apache/ambari/server/state/kerberos/AbstractKerberosDescriptorContainer.java

@@ -24,9 +24,11 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * AbstractKerberosDescriptorContainer is an abstract class implementing AbstractKerberosDescriptor
@@ -64,6 +66,14 @@ import java.util.Map;
  *            "title": "KerberosConfigurationDescriptor"
  *            "type": "{@link org.apache.ambari.server.state.kerberos.KerberosConfigurationDescriptor}"
  *          }
+ *        },
+ *        "auth_to_local": {
+ *          "description": "A list of configuration properties declaring which properties are auth-to-local values
+ *          "type": "array",
+ *          "items": {
+ *            "title": "String"
+ *            "type": "{@link java.lang.String}"
+ *          }
  *        }
  *      }
  *   }
@@ -85,6 +95,12 @@ public abstract class AbstractKerberosDescriptorContainer extends AbstractKerber
    */
   private Map<String, KerberosConfigurationDescriptor> configurations = null;
 
+  /**
+   * A Set of configuration identifiers (config-type/property_name) that indicate which properties
+   * contain auth_to_local values.
+   */
+  private Set<String> authToLocalProperties = null;
+
   /**
    * Constructs a new AbstractKerberosDescriptorContainer
    * <p/>
@@ -115,6 +131,16 @@ public abstract class AbstractKerberosDescriptorContainer extends AbstractKerber
           }
         }
       }
+
+      // (Safely) Get the set of KerberosConfigurationDescriptors
+      list = data.get(KerberosDescriptorType.AUTH_TO_LOCAL_PROPERTY.getDescriptorPluralName());
+      if (list instanceof Collection) {
+        for (Object item : (Collection) list) {
+          if (item instanceof String) {
+            putAuthToLocalProperty((String) item);
+          }
+        }
+      }
     }
   }
 
@@ -397,6 +423,33 @@ public abstract class AbstractKerberosDescriptorContainer extends AbstractKerber
     return ((name == null) || (configurations == null)) ? null : configurations.get(name);
   }
 
+  /**
+   * Adds the specified property name to the set of <code>auth_to_local</code> property names.
+   * <p/>
+   * Each <code>auth_to_local</code> property name is expected to be in the following format:
+   * <code>config-type/property_name</code>`
+   *
+   * @param authToLocalProperty the auth_to_local property to add
+   */
+  public void putAuthToLocalProperty(String authToLocalProperty) {
+    if (authToLocalProperty != null) {
+      if (authToLocalProperties == null) {
+        authToLocalProperties = new HashSet<String>();
+      }
+
+      authToLocalProperties.add(authToLocalProperty);
+    }
+  }
+
+  /**
+   * Gets the set of <code>auth_to_local</code> property names.
+   *
+   * @return a Set of String values; or null if not set
+   */
+  public Set<String> getAuthToLocalProperties() {
+    return authToLocalProperties;
+  }
+
   /**
    * Test this AbstractKerberosDescriptor to see if it is a container.
    * <p/>
@@ -459,6 +512,13 @@ public abstract class AbstractKerberosDescriptorContainer extends AbstractKerber
           }
         }
       }
+
+      Set<String> updatedAuthToLocalProperties = updates.getAuthToLocalProperties();
+      if (updatedAuthToLocalProperties != null) {
+        for (String updatedAuthToLocalProperty : updatedAuthToLocalProperties) {
+          putAuthToLocalProperty(updatedAuthToLocalProperty);
+        }
+      }
     }
   }
 
@@ -586,6 +646,11 @@ public abstract class AbstractKerberosDescriptorContainer extends AbstractKerber
       map.put(KerberosDescriptorType.CONFIGURATION.getDescriptorPluralName(), list);
     }
 
+    if (authToLocalProperties != null) {
+      List<String> list = new ArrayList<String>(authToLocalProperties);
+      map.put(KerberosDescriptorType.AUTH_TO_LOCAL_PROPERTY.getDescriptorPluralName(), list);
+    }
+
     return map;
   }
 

ambari-server/src/main/java/org/apache/ambari/server/state/kerberos/KerberosDescriptorType.java

@@ -24,7 +24,8 @@ public enum KerberosDescriptorType {
   IDENTITY("identity", "identities"),
   PRINCIPAL("principal", "principals"),
   KEYTAB("keytab", "keytabs"),
-  CONFIGURATION("configuration", "configurations");
+  CONFIGURATION("configuration", "configurations"),
+  AUTH_TO_LOCAL_PROPERTY("auth_to_local_property", "auth_to_local_properties");
 
   private final String descriptorName;
   private final String descriptorPluralName;

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/kerberos.json

@@ -19,13 +19,16 @@
           "name": "/hdfs"
         }
       ],
+      "auth_to_local_properties" : [
+        "core-site/hadoop.security.auth_to_local"
+      ],
       "configurations": [
         {
           "core-site": {
             "hadoop.security.authentication": "kerberos",
             "hadoop.rpc.protection": "authentication",
             "hadoop.security.authorization": "true",
-            "hadoop.security.auth_to_local": "_AUTH_TO_LOCAL_RULES",
+            "hadoop.security.auth_to_local": "",
             "hadoop.http.authentication.kerberos.name.rules": "",
             "hadoop.http.filter.initializers": "",
             "hadoop.http.authentication.type": "simple",

ambari-server/src/main/resources/common-services/KERBEROS/1.10.3-10/package/scripts/kerberos_common.py

@@ -395,6 +395,11 @@ class KerberosScript(Script):
 
             principal = get_property_value(item, 'principal')
             if principal is not None:
-              self.put_structured_out({
-                principal.replace("_HOST", params.hostname): keytab_file_path
-              })
+              curr_content = Script.structuredOut
+
+              if "keytabs" not in curr_content:
+                curr_content['keytabs'] = {}
+
+              curr_content['keytabs'][principal.replace("_HOST", params.hostname)] = keytab_file_path
+
+              self.put_structured_out(curr_content)

ambari-server/src/main/resources/common-services/OOZIE/4.0.0.2.0/kerberos.json

@@ -13,6 +13,9 @@
           "name": "/hdfs"
         }
       ],
+      "auth_to_local_properties" : [
+        "oozie-site/oozie.authentication.kerberos.name.rules"
+      ],
       "configurations": [
         {
           "oozie-site": {
@@ -20,7 +23,7 @@
             "oozie.service.AuthorizationService.authorization.enabled": "true",
             "oozie.service.HadoopAccessorService.kerberos.enabled": "true",
             "local.realm": "${realm}",
-            "oozie.authentication.kerberos.name.rules": "_AUTH_TO_LOCAL_RULES"
+            "oozie.authentication.kerberos.name.rules": ""
           }
         }
       ],

ambari-server/src/test/java/org/apache/ambari/server/agent/TestHeartbeatHandler.java

@@ -2521,6 +2521,7 @@ public class TestHeartbeatHandler {
     executionCommand.setHostLevelParams(hlp);
 
     Map<String, String> commandparams = new HashMap<String, String>();
+    commandparams.put(KerberosServerAction.AUTHENTICATED_USER_NAME, "admin");
     commandparams.put(KerberosServerAction.DATA_DIRECTORY, createTestKeytabData().getAbsolutePath());
     executionCommand.setCommandParams(commandparams);
 

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

@@ -3127,9 +3127,9 @@ public class AmbariManagementControllerTest {
     // sch5 to start
     Stage stage1 = null, stage2 = null, stage3 = null;
     for (Stage s : stages) {
-      if (s.getStageId() == 1) { stage1 = s; }
-      if (s.getStageId() == 2) { stage2 = s; }
-      if (s.getStageId() == 3) { stage3 = s; }
+      if (s.getStageId() == 0) { stage1 = s; }
+      if (s.getStageId() == 1) { stage2 = s; }
+      if (s.getStageId() == 2) { stage3 = s; }
     }
 
     Assert.assertEquals(2, stage1.getExecutionCommands(host1).size());

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

@@ -236,4 +236,25 @@ public class AuthToLocalBuilderTest {
         "DEFAULT",
         builder.generate("EXAMPLE.COM"));
   }
+
+  @Test
+  public void testCopy() {
+    AuthToLocalBuilder builder = new AuthToLocalBuilder();
+
+    builder.addRule("nn/_HOST@EXAMPLE.COM", "hdfs");
+    builder.addRule("dn/_HOST@EXAMPLE.COM", "hdfs");
+    builder.addRule("jn/_HOST@EXAMPLE.COM", "hdfs");
+    builder.addRule("rm/_HOST@EXAMPLE.COM", "yarn");
+    builder.addRule("jhs/_HOST@EXAMPLE.COM", "mapred");
+    builder.addRule("hm/_HOST@EXAMPLE.COM", "hbase");
+    builder.addRule("rs/_HOST@EXAMPLE.COM", "hbase");
+
+    builder.addRule("foobar@EXAMPLE.COM", "hdfs");
+
+    AuthToLocalBuilder copy = builder.copy();
+
+    assertNotSame(builder, copy);
+    assertEquals(copy.generate("EXAMPLE.COM"), builder.generate("EXAMPLE.COM"));
+
+  }
 }

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

@@ -375,7 +375,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(service1.getName()).andReturn("SERVICE1").anyTimes();
     expect(service1.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
     service1.setSecurityState(SecurityState.SECURED_KERBEROS);
     expectLastCall().once();
 
@@ -383,7 +383,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(service2.getName()).andReturn("SERVICE2").anyTimes();
     expect(service2.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
     service2.setSecurityState(SecurityState.SECURED_KERBEROS);
     expectLastCall().once();
 
@@ -448,6 +448,9 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))
         .andReturn(Collections.<String, Map<String, String>>emptyMap())
         .once();
+    expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, null))
+        .andReturn(Collections.<String, Map<String, String>>emptyMap())
+        .once();
     expect(ambariManagementController.getRoleCommandOrder(cluster))
         .andReturn(createNiceMock(RoleCommandOrder.class))
         .once();
@@ -461,10 +464,7 @@ public class KerberosHelperTest extends EasyMockSupport {
             }});
           }
         })
-        .once();
-    expect(configHelper.getEffectiveConfigAttributes(anyObject(Cluster.class), anyObject(Map.class)))
-        .andReturn(Collections.<String, Map<String, Map<String, String>>>emptyMap())
-        .once();
+        .times(2);
 
     final KerberosPrincipalDescriptor principalDescriptor1 = createNiceMock(KerberosPrincipalDescriptor.class);
     expect(principalDescriptor1.getValue()).andReturn("component1/_HOST@${realm}").once();
@@ -625,7 +625,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(service1.getName()).andReturn("SERVICE1").anyTimes();
     expect(service1.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
     service1.setSecurityState(SecurityState.UNSECURED);
     expectLastCall().once();
 
@@ -633,7 +633,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(service2.getName()).andReturn("SERVICE2").anyTimes();
     expect(service2.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
     service2.setSecurityState(SecurityState.UNSECURED);
     expectLastCall().once();
 
@@ -694,6 +694,9 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))
         .andReturn(Collections.<String, Map<String, String>>emptyMap())
         .once();
+    expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, null))
+        .andReturn(Collections.<String, Map<String, String>>emptyMap())
+        .once();
     expect(ambariManagementController.getRoleCommandOrder(cluster))
         .andReturn(createNiceMock(RoleCommandOrder.class))
         .once();
@@ -707,10 +710,7 @@ public class KerberosHelperTest extends EasyMockSupport {
             }});
           }
         })
-        .once();
-    expect(configHelper.getEffectiveConfigAttributes(anyObject(Cluster.class), anyObject(Map.class)))
-        .andReturn(Collections.<String, Map<String, Map<String, String>>>emptyMap())
-        .once();
+        .times(2);
 
     final KerberosPrincipalDescriptor principalDescriptor1 = createNiceMock(KerberosPrincipalDescriptor.class);
     expect(principalDescriptor1.getValue()).andReturn("component1/_HOST@${realm}").once();
@@ -852,13 +852,13 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(service1.getName()).andReturn("SERVICE1").anyTimes();
     expect(service1.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
 
     final Service service2 = createStrictMock(Service.class);
     expect(service2.getName()).andReturn("SERVICE2").anyTimes();
     expect(service2.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
 
     final Map<String, String> kerberosEnvProperties = createNiceMock(Map.class);
     expect(kerberosEnvProperties.get("kdc_type")).andReturn("mit-kdc").anyTimes();
@@ -920,6 +920,9 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))
         .andReturn(Collections.<String, Map<String, String>>emptyMap())
         .once();
+    expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, null))
+        .andReturn(Collections.<String, Map<String, String>>emptyMap())
+        .once();
     expect(ambariManagementController.getRoleCommandOrder(cluster))
         .andReturn(createNiceMock(RoleCommandOrder.class))
         .once();
@@ -933,10 +936,7 @@ public class KerberosHelperTest extends EasyMockSupport {
             }});
           }
         })
-        .once();
-    expect(configHelper.getEffectiveConfigAttributes(anyObject(Cluster.class), anyObject(Map.class)))
-        .andReturn(Collections.<String, Map<String, Map<String, String>>>emptyMap())
-        .once();
+        .times(2);
 
     final KerberosPrincipalDescriptor principalDescriptor1 = createNiceMock(KerberosPrincipalDescriptor.class);
     expect(principalDescriptor1.getValue()).andReturn("component1/_HOST@${realm}").once();
@@ -1130,13 +1130,13 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(service1.getName()).andReturn("SERVICE1").anyTimes();
     expect(service1.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
 
     final Service service2 = createStrictMock(Service.class);
     expect(service2.getName()).andReturn("SERVICE2").anyTimes();
     expect(service2.getServiceComponents())
         .andReturn(Collections.<String, ServiceComponent>emptyMap())
-        .times(2);
+        .times(3);
 
     final Map<String, String> kerberosEnvProperties = createNiceMock(Map.class);
     expect(kerberosEnvProperties.get("kdc_type")).andReturn("mit-kdc").anyTimes();
@@ -1198,6 +1198,9 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))
         .andReturn(Collections.<String, Map<String, String>>emptyMap())
         .once();
+    expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, null))
+        .andReturn(Collections.<String, Map<String, String>>emptyMap())
+        .once();
     expect(ambariManagementController.getRoleCommandOrder(cluster))
         .andReturn(createNiceMock(RoleCommandOrder.class))
         .once();
@@ -1211,16 +1214,19 @@ public class KerberosHelperTest extends EasyMockSupport {
             }});
           }
         })
-        .once();
-    expect(configHelper.getEffectiveConfigAttributes(anyObject(Cluster.class), anyObject(Map.class)))
-        .andReturn(Collections.<String, Map<String, Map<String, String>>>emptyMap())
-        .once();
+        .times(2);
+
+    final KerberosPrincipalDescriptor principalDescriptor1a = createMock(KerberosPrincipalDescriptor.class);
+    expect(principalDescriptor1a.getValue()).andReturn("component1a/_HOST@${realm}").anyTimes();
+    expect(principalDescriptor1a.getType()).andReturn(KerberosPrincipalType.SERVICE).anyTimes();
+    expect(principalDescriptor1a.getLocalUsername()).andReturn(null).anyTimes();
+    expect(principalDescriptor1a.getConfiguration()).andReturn("service1b-site/component1.kerberos.principal").anyTimes();
 
-    final KerberosPrincipalDescriptor principalDescriptor1 = createMock(KerberosPrincipalDescriptor.class);
-    expect(principalDescriptor1.getValue()).andReturn("component1/_HOST@${realm}").anyTimes();
-    expect(principalDescriptor1.getType()).andReturn(KerberosPrincipalType.SERVICE).anyTimes();
-    expect(principalDescriptor1.getLocalUsername()).andReturn(null).anyTimes();
-    expect(principalDescriptor1.getConfiguration()).andReturn("service1-site/component1.kerberos.principal").anyTimes();
+    final KerberosPrincipalDescriptor principalDescriptor1b = createMock(KerberosPrincipalDescriptor.class);
+    expect(principalDescriptor1b.getValue()).andReturn("component1b/_HOST@${realm}").anyTimes();
+    expect(principalDescriptor1b.getType()).andReturn(KerberosPrincipalType.SERVICE).anyTimes();
+    expect(principalDescriptor1b.getLocalUsername()).andReturn(null).anyTimes();
+    expect(principalDescriptor1b.getConfiguration()).andReturn("service1b-site/component1.kerberos.principal").anyTimes();
 
     final KerberosPrincipalDescriptor principalDescriptor3 = createMock(KerberosPrincipalDescriptor.class);
     expect(principalDescriptor3.getValue()).andReturn("component3/${host}@${realm}").anyTimes();
@@ -1246,44 +1252,61 @@ public class KerberosHelperTest extends EasyMockSupport {
 
     final KerberosIdentityDescriptor identityDescriptor1a = createMock(KerberosIdentityDescriptor.class);
     expect(identityDescriptor1a.getName()).andReturn("identity1a").anyTimes();
-    expect(identityDescriptor1a.getPrincipalDescriptor()).andReturn(principalDescriptor1).anyTimes();
+    expect(identityDescriptor1a.getPrincipalDescriptor()).andReturn(principalDescriptor1a).anyTimes();
     expect(identityDescriptor1a.getKeytabDescriptor()).andReturn(keytabDescriptor1).anyTimes();
 
     final KerberosIdentityDescriptor identityDescriptor1b = createMock(KerberosIdentityDescriptor.class);
     expect(identityDescriptor1b.getName()).andReturn("identity1b").anyTimes();
+    expect(identityDescriptor1b.getPrincipalDescriptor()).andReturn(principalDescriptor1b).anyTimes();
 
     final KerberosIdentityDescriptor identityDescriptor3 = createMock(KerberosIdentityDescriptor.class);
     expect(identityDescriptor3.getName()).andReturn("identity3").anyTimes();
     expect(identityDescriptor3.getPrincipalDescriptor()).andReturn(principalDescriptor3).anyTimes();
     expect(identityDescriptor3.getKeytabDescriptor()).andReturn(keytabDescriptor3).anyTimes();
 
+    final ArrayList<KerberosIdentityDescriptor> identityDescriptors1 = new ArrayList<KerberosIdentityDescriptor>() {{
+      add(identityDescriptor1a);
+      add(identityDescriptor1b);
+    }};
     final KerberosComponentDescriptor componentDescriptor1 = createStrictMock(KerberosComponentDescriptor.class);
-    expect(componentDescriptor1.getIdentities(true)).
-        andReturn(new ArrayList<KerberosIdentityDescriptor>() {{
-          add(identityDescriptor1a);
-          add(identityDescriptor1b);
-        }}).once();
-    expect(componentDescriptor1.getConfigurations(true)).andReturn(null).once();
-
+    expect(componentDescriptor1.getIdentities(true)).andReturn(identityDescriptors1).times(1);
+    expect(componentDescriptor1.getConfigurations(true)).andReturn(null).times(1);
+    expect(componentDescriptor1.getIdentities(true)).andReturn(identityDescriptors1).times(1);
+    expect(componentDescriptor1.getAuthToLocalProperties()).andReturn(null).times(1);
+
+    final ArrayList<KerberosIdentityDescriptor> identityDescriptors3 = new ArrayList<KerberosIdentityDescriptor>() {{
+      add(identityDescriptor3);
+    }};
     final KerberosComponentDescriptor componentDescriptor3 = createStrictMock(KerberosComponentDescriptor.class);
-    expect(componentDescriptor3.getIdentities(true)).
-        andReturn(new ArrayList<KerberosIdentityDescriptor>() {{
-          add(identityDescriptor3);
-        }}).once();
-    expect(componentDescriptor3.getConfigurations(true)).andReturn(null).once();
+    expect(componentDescriptor3.getIdentities(true)).andReturn(identityDescriptors3).times(1);
+    expect(componentDescriptor3.getConfigurations(true)).andReturn(null).times(1);
 
     final KerberosServiceDescriptor serviceDescriptor1 = createMock(KerberosServiceDescriptor.class);
-    expect(serviceDescriptor1.getIdentities(true)).andReturn(null).once();
+    expect(serviceDescriptor1.getIdentities(true)).andReturn(null).times(1);
+    expect(serviceDescriptor1.getName()).andReturn("SERVICE1").times(1);
+    expect(serviceDescriptor1.getIdentities(true)).andReturn(null).times(1);
+    expect(serviceDescriptor1.getComponents()).andReturn(new HashMap<String, KerberosComponentDescriptor>(){{
+      put("COMPONENT1", componentDescriptor1);
+    }}).times(1);
     expect(serviceDescriptor1.getComponent("COMPONENT1")).andReturn(componentDescriptor1).once();
+    expect(serviceDescriptor1.getAuthToLocalProperties()).andReturn(null).once();
 
     final KerberosServiceDescriptor serviceDescriptor3 = createMock(KerberosServiceDescriptor.class);
-    expect(serviceDescriptor3.getIdentities(true)).andReturn(null).once();
+    expect(serviceDescriptor3.getIdentities(true)).andReturn(null).times(1);
+    expect(serviceDescriptor3.getName()).andReturn("SERVICE3").times(1);
     expect(serviceDescriptor3.getComponent("COMPONENT3")).andReturn(componentDescriptor3).once();
 
     final KerberosDescriptor kerberosDescriptor = createStrictMock(KerberosDescriptor.class);
     expect(kerberosDescriptor.getProperties()).andReturn(null).once();
     expect(kerberosDescriptor.getService("SERVICE1")).andReturn(serviceDescriptor1).once();
     expect(kerberosDescriptor.getService("SERVICE3")).andReturn(serviceDescriptor3).once();
+    expect(kerberosDescriptor.getIdentities()).andReturn(null).once();
+    expect(kerberosDescriptor.getAuthToLocalProperties()).andReturn(null).once();
+    expect(kerberosDescriptor.getServices()).andReturn(new HashMap<String, KerberosServiceDescriptor>()
+    {{
+      put("SERVCE1", serviceDescriptor1);
+      put("SERVCE2", serviceDescriptor3);
+    }}).once();
 
     setupGetDescriptorFromCluster(kerberosDescriptor);
 

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

@@ -20,7 +20,6 @@ package org.apache.ambari.server.controller.internal;
 
 import com.google.inject.Injector;
 import org.apache.ambari.server.controller.AmbariManagementController;
-import org.apache.ambari.server.controller.KerberosHelper;
 import org.apache.ambari.server.controller.MaintenanceStateHelper;
 import org.apache.ambari.server.controller.RequestStatusResponse;
 import org.apache.ambari.server.controller.ResourceProviderFactory;
@@ -35,7 +34,6 @@ import org.apache.ambari.server.controller.utilities.PredicateBuilder;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
-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;
@@ -60,7 +58,6 @@ import java.util.Set;
 import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.createNiceMock;
-import static org.easymock.EasyMock.createStrictMock;
 import static org.easymock.EasyMock.eq;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.replay;
@@ -304,7 +301,7 @@ public class HostComponentResourceProviderTest {
     HostComponentResourceProvider provider =
         new TestHostComponentResourceProvider(PropertyHelper.getPropertyIds(type),
             PropertyHelper.getKeyPropertyIds(type),
-            managementController, injector, maintenanceStateHelper, null);
+            managementController, injector, maintenanceStateHelper);
 
     expect(resourceProviderFactory.getHostComponentResourceProvider(anyObject(Set.class),
         anyObject(Map.class),
@@ -355,8 +352,6 @@ public class HostComponentResourceProviderTest {
     ServiceComponentHost clientComponentHost = createNiceMock(ServiceComponentHost.class);
     RequestStageContainer stageContainer = createNiceMock(RequestStageContainer.class);
     MaintenanceStateHelper maintenanceStateHelper = createNiceMock(MaintenanceStateHelper.class);
-    // INIT->INSTALLED state transition causes check for kerberized cluster
-    KerberosHelper kerberosHelper = createStrictMock(KerberosHelper.class);
 
     Collection<String> hosts = new HashSet<String>();
     hosts.add("Host100");
@@ -432,18 +427,16 @@ public class HostComponentResourceProviderTest {
     HostComponentResourceProvider provider =
         new TestHostComponentResourceProvider(PropertyHelper.getPropertyIds(type),
             PropertyHelper.getKeyPropertyIds(type),
-            managementController, injector, maintenanceStateHelper, kerberosHelper);
+            managementController, injector, maintenanceStateHelper);
 
     expect(resourceProviderFactory.getHostComponentResourceProvider(anyObject(Set.class),
         anyObject(Map.class),
         eq(managementController))).
         andReturn(provider).anyTimes();
 
-    expect(kerberosHelper.isClusterKerberosEnabled(cluster)).andReturn(false).times(2);
-
     // replay
     replay(managementController, response, resourceProviderFactory, clusters, cluster, service,
-        component, componentHost, stageContainer, maintenanceStateHelper, kerberosHelper, clientComponent,
+        component, componentHost, stageContainer, maintenanceStateHelper, clientComponent,
         clientComponentHost);
 
     Map<String, Object> properties = new LinkedHashMap<String, Object>();
@@ -453,109 +446,8 @@ public class HostComponentResourceProviderTest {
 
     assertSame(response, requestResponse);
     // verify
-    verify(managementController, response, resourceProviderFactory, stageContainer, kerberosHelper,
-           clientComponent, clientComponentHost);
-  }
-
-  @Test
-  public void testInstallAndStart_kerberizedCluster() throws Exception {
-    Resource.Type type = Resource.Type.HostComponent;
-
-    AmbariManagementController managementController = createMock(AmbariManagementController.class);
-    RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
-    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
-    Injector injector = createNiceMock(Injector.class);
-    Clusters clusters = createNiceMock(Clusters.class);
-    Cluster cluster = createNiceMock(Cluster.class);
-    Service service = createNiceMock(Service.class);
-    ServiceComponent component = createNiceMock(ServiceComponent.class);
-    ServiceComponentHost componentHost = createNiceMock(ServiceComponentHost.class);
-    RequestStageContainer stageContainer = createNiceMock(RequestStageContainer.class);
-    MaintenanceStateHelper maintenanceStateHelper = createNiceMock(MaintenanceStateHelper.class);
-    KerberosHelper kerberosHelper = createStrictMock(KerberosHelper.class);
-
-    Collection<String> hosts = new HashSet<String>();
-    hosts.add("Host100");
-
-    Map<String, String> mapRequestProps = new HashMap<String, String>();
-    mapRequestProps.put("context", "Install and start components on added hosts");
-
-    Set<ServiceComponentHostResponse> nameResponse = new HashSet<ServiceComponentHostResponse>();
-    nameResponse.add(new ServiceComponentHostResponse(
-        "Cluster102", "Service100", "Component100", "Host100", "INIT", "", "INIT", "", null));
-    Set<ServiceComponentHostResponse> nameResponse2 = new HashSet<ServiceComponentHostResponse>();
-    nameResponse2.add(new ServiceComponentHostResponse(
-        "Cluster102", "Service100", "Component100", "Host100", "INIT", "", "INSTALLED", "", null));
-
-
-    // set expectations
-    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
-    expect(managementController.findServiceName(cluster, "Component100")).andReturn("Service100").anyTimes();
-    expect(clusters.getCluster("Cluster102")).andReturn(cluster).anyTimes();
-    expect(cluster.getService("Service100")).andReturn(service).anyTimes();
-    expect(service.getServiceComponent("Component100")).andReturn(component).anyTimes();
-    expect(component.getServiceComponentHost("Host100")).andReturn(componentHost).anyTimes();
-    expect(component.getName()).andReturn("Component100").anyTimes();
-    // actual state is always INIT until stages actually execute
-    expect(componentHost.getState()).andReturn(State.INIT).anyTimes();
-    expect(componentHost.getHostName()).andReturn("Host100").anyTimes();
-    expect(componentHost.getServiceComponentName()).andReturn("Component100").anyTimes();
-    expect(response.getMessage()).andReturn("response msg").anyTimes();
-
-    //Cluster is default type.  Maintenance mode is not being tested here so the default is returned.
-    expect(maintenanceStateHelper.isOperationAllowed(Resource.Type.Cluster, componentHost)).andReturn(true).anyTimes();
-
-    expect(managementController.getHostComponents(
-        EasyMock.<Set<ServiceComponentHostRequest>>anyObject())).andReturn(nameResponse);
-    expect(managementController.getHostComponents(
-        EasyMock.<Set<ServiceComponentHostRequest>>anyObject())).andReturn(nameResponse2);
-
-    Map<String, Map<State, List<ServiceComponentHost>>> changedHosts =
-        new HashMap<String, Map<State, List<ServiceComponentHost>>>();
-    List<ServiceComponentHost> changedComponentHosts = Collections.singletonList(componentHost);
-    changedHosts.put("Component100", Collections.singletonMap(State.INSTALLED, changedComponentHosts));
-
-    Map<String, Map<State, List<ServiceComponentHost>>> changedHosts2 =
-        new HashMap<String, Map<State, List<ServiceComponentHost>>>();
-    List<ServiceComponentHost> changedComponentHosts2 = Collections.singletonList(componentHost);
-    changedHosts2.put("Component100", Collections.singletonMap(State.STARTED, changedComponentHosts2));
-
-    expect(managementController.addStages(null, cluster, mapRequestProps, null, null, null, changedHosts,
-        Collections.<ServiceComponentHost>emptyList(), false, false)).andReturn(stageContainer).once();
-
-    expect(managementController.addStages(stageContainer, cluster, mapRequestProps, null, null, null, changedHosts2,
-        Collections.<ServiceComponentHost>emptyList(), false, false)).andReturn(stageContainer).once();
-
-    stageContainer.persist();
-    expect(stageContainer.getProjectedState("Host100", "Component100")).andReturn(State.INSTALLED).once();
-    expect(stageContainer.getRequestStatusResponse()).andReturn(response).once();
-
-    HostComponentResourceProvider provider =
-        new TestHostComponentResourceProvider(PropertyHelper.getPropertyIds(type),
-            PropertyHelper.getKeyPropertyIds(type),
-            managementController, injector, maintenanceStateHelper, kerberosHelper);
-
-    expect(resourceProviderFactory.getHostComponentResourceProvider(anyObject(Set.class),
-        anyObject(Map.class),
-        eq(managementController))).
-        andReturn(provider).anyTimes();
-
-    expect(kerberosHelper.isClusterKerberosEnabled(cluster)).andReturn(true).once();
-    expect(kerberosHelper.toggleKerberos(cluster, SecurityType.KERBEROS, stageContainer)).
-        andReturn(stageContainer).once();
-
-    // replay
-    replay(managementController, response, resourceProviderFactory, clusters, cluster, service,
-        component, componentHost, stageContainer, maintenanceStateHelper, kerberosHelper);
-
-    Map<String, Object> properties = new LinkedHashMap<String, Object>();
-    properties.put(HostComponentResourceProvider.HOST_COMPONENT_STATE_PROPERTY_ID, "STARTED");
-
-    RequestStatusResponse requestResponse = provider.installAndStart("Cluster102", hosts);
-
-    assertSame(response, requestResponse);
-    // verify
-    verify(managementController, response, resourceProviderFactory, stageContainer, kerberosHelper);
+    verify(managementController, response, resourceProviderFactory, stageContainer,
+        clientComponent, clientComponentHost);
   }
 
   @Test
@@ -657,8 +549,8 @@ public class HostComponentResourceProviderTest {
     HostComponentResourceProvider provider =
         new TestHostComponentResourceProvider(PropertyHelper.getPropertyIds(type),
             PropertyHelper.getKeyPropertyIds(type),
-            controller, injector, injector.getInstance(MaintenanceStateHelper.class),
-            injector.getInstance(KerberosHelper.class));
+            controller, injector, injector.getInstance(MaintenanceStateHelper.class)
+        );
     RequestStageContainer requestStages = provider.updateHostComponents(null, requests, requestProperties, runSmokeTest);
     requestStages.persist();
     return requestStages.getRequestStatusResponse();
@@ -675,8 +567,7 @@ public class HostComponentResourceProviderTest {
      */
     public TestHostComponentResourceProvider(Set<String> propertyIds, Map<Resource.Type, String> keyPropertyIds,
                                              AmbariManagementController managementController, Injector injector,
-                                             MaintenanceStateHelper maintenanceStateHelper,
-                                             KerberosHelper kerberosHelper) throws Exception {
+                                             MaintenanceStateHelper maintenanceStateHelper) throws Exception {
 
       super(propertyIds, keyPropertyIds, managementController, injector);
 
@@ -685,10 +576,6 @@ public class HostComponentResourceProviderTest {
       Field f = c.getDeclaredField("maintenanceStateHelper");
       f.setAccessible(true);
       f.set(this, maintenanceStateHelper);
-
-      f = c.getDeclaredField("kerberosHelper");
-      f.setAccessible(true);
-      f.set(this, kerberosHelper);
     }
   }
 }

ambari-server/src/test/java/org/apache/ambari/server/serveraction/kerberos/UpdateKerberosConfigsServerActionTest.java

@@ -26,6 +26,7 @@ import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.controller.ConfigurationRequest;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.easymock.EasyMock;
 import org.junit.After;
@@ -50,9 +51,11 @@ public class UpdateKerberosConfigsServerActionTest {
   String dataDir;
   private Injector injector;
   private UpdateKerberosConfigsServerAction action;
-  final AmbariManagementController controller = EasyMock.createNiceMock(AmbariManagementController.class);
-  Clusters clusters = EasyMock.createNiceMock(Clusters.class);
-  Cluster cluster = EasyMock.createNiceMock(Cluster.class);
+
+  private final AmbariManagementController controller = EasyMock.createNiceMock(AmbariManagementController.class);
+  private final ConfigHelper configHelper = createNiceMock(ConfigHelper.class);
+  private final Clusters clusters = EasyMock.createNiceMock(Clusters.class);
+  private final Cluster cluster = EasyMock.createNiceMock(Cluster.class);
 
   @Before
   public void setup() throws Exception {
@@ -60,14 +63,13 @@ public class UpdateKerberosConfigsServerActionTest {
     setupConfigDat();
 
     expect(controller.getClusters()).andReturn(clusters).once();
-
-    expect(controller.createConfiguration(anyObject(ConfigurationRequest.class))).andReturn(
-      null).once();
-
     replay(controller);
 
-    expect(cluster.getConfigsByType("hdfs-site")).andReturn(null).once();
-    expect(cluster.getDesiredConfigByType("hdfs-site")).andReturn(null).once();
+    configHelper.updateConfigType(anyObject(Cluster.class), anyObject(AmbariManagementController.class),
+        anyObject(String.class), anyObject(Map.class), anyObject(String.class), anyObject(String.class));
+    expectLastCall().atLeastOnce();
+    replay(configHelper);
+
     replay(cluster);
 
     expect(clusters.getCluster(anyObject(String.class))).andReturn(cluster).once();
@@ -78,6 +80,7 @@ public class UpdateKerberosConfigsServerActionTest {
       @Override
       protected void configure() {
         bind(AmbariManagementController.class).toInstance(controller);
+        bind(ConfigHelper.class).toInstance(configHelper);
       }
     });
     action = injector.getInstance(UpdateKerberosConfigsServerAction.class);
@@ -137,9 +140,7 @@ public class UpdateKerberosConfigsServerActionTest {
 
     action.execute(requestSharedDataContext);
 
-    verify(controller);
-    verify(clusters);
-    verify(cluster);
+    verify(controller, clusters, cluster, configHelper);
   }
 
   @Test

ambari-server/src/test/java/org/apache/ambari/server/state/kerberos/KerberosComponentDescriptorTest.java

@@ -25,8 +25,11 @@ import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
 
 public class KerberosComponentDescriptorTest {
   public static final String JSON_VALUE =
@@ -42,7 +45,10 @@ public class KerberosComponentDescriptorTest {
           "        \"service.component.property2\": \"value2\"" +
           "      }" +
           "    }" +
-          "  ]" +
+          "  ]," +
+          "  \"auth_to_local_properties\": [" +
+          "      component.name.rules1" +
+          "    ]" +
           "}";
 
   public static final Map<String, Object> MAP_VALUE =
@@ -66,6 +72,9 @@ public class KerberosComponentDescriptorTest {
               }
             });
           }});
+          put(KerberosDescriptorType.AUTH_TO_LOCAL_PROPERTY.getDescriptorPluralName(), new ArrayList<String>() {{
+            add("component.name.rules2");
+          }});
         }
       };
 
@@ -96,6 +105,11 @@ public class KerberosComponentDescriptorTest {
     Assert.assertEquals(2, properties.size());
     Assert.assertEquals("value1", properties.get("service.component.property1"));
     Assert.assertEquals("value2", properties.get("service.component.property2"));
+
+    Set<String> authToLocalProperties = componentDescriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(1, authToLocalProperties.size());
+    Assert.assertEquals("component.name.rules1", authToLocalProperties.iterator().next());
   }
 
   public static void validateFromMap(KerberosComponentDescriptor componentDescriptor) {
@@ -125,6 +139,11 @@ public class KerberosComponentDescriptorTest {
     Assert.assertEquals(2, properties.size());
     Assert.assertEquals("red", properties.get("service.component.property1"));
     Assert.assertEquals("green", properties.get("service.component.property"));
+
+    Set<String> authToLocalProperties = componentDescriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(1, authToLocalProperties.size());
+    Assert.assertEquals("component.name.rules2", authToLocalProperties.iterator().next());
   }
 
   public static void validateUpdatedData(KerberosComponentDescriptor componentDescriptor) {
@@ -154,6 +173,15 @@ public class KerberosComponentDescriptorTest {
     Assert.assertEquals("red", properties.get("service.component.property1"));
     Assert.assertEquals("value2", properties.get("service.component.property2"));
     Assert.assertEquals("green", properties.get("service.component.property"));
+
+    Set<String> authToLocalProperties = componentDescriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(2, authToLocalProperties.size());
+    // guarantee ordering...
+    Iterator<String> iterator = new TreeSet<String>(authToLocalProperties).iterator();
+    Assert.assertEquals("component.name.rules1", iterator.next());
+    Assert.assertEquals("component.name.rules2", iterator.next());
+
   }
 
   private static KerberosComponentDescriptor createFromJSON() {

ambari-server/src/test/java/org/apache/ambari/server/state/kerberos/KerberosDescriptorTest.java

@@ -24,8 +24,11 @@ import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
 
 public class KerberosDescriptorTest {
   private static final KerberosDescriptorFactory KERBEROS_DESCRIPTOR_FACTORY = new KerberosDescriptorFactory();
@@ -37,6 +40,9 @@ public class KerberosDescriptorTest {
           "      \"realm\": \"${cluster-env/kerberos_domain}\"," +
           "      \"keytab_dir\": \"/etc/security/keytabs\"" +
           "    }," +
+          "  \"auth_to_local_properties\": [" +
+          "      generic.name.rules" +
+          "    ]," +
           "  \"services\": [" +
           KerberosServiceDescriptorTest.JSON_VALUE +
           "    ]" +
@@ -50,6 +56,10 @@ public class KerberosDescriptorTest {
             put("some.property", "Hello World");
           }});
 
+          put(KerberosDescriptorType.AUTH_TO_LOCAL_PROPERTY.getDescriptorPluralName(), new ArrayList<String>() {{
+            add("global.name.rules");
+          }});
+
           put(KerberosDescriptorType.SERVICE.getDescriptorPluralName(), new ArrayList<Object>() {{
             add(KerberosServiceDescriptorTest.MAP_VALUE);
           }});
@@ -102,6 +112,11 @@ public class KerberosDescriptorTest {
     Assert.assertEquals("${cluster-env/kerberos_domain}", properties.get("realm"));
     Assert.assertEquals("/etc/security/keytabs", properties.get("keytab_dir"));
 
+    Set<String> authToLocalProperties = descriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(1, authToLocalProperties.size());
+    Assert.assertEquals("generic.name.rules", authToLocalProperties.iterator().next());
+
     Map<String, KerberosServiceDescriptor> serviceDescriptors = descriptor.getServices();
     Assert.assertNotNull(serviceDescriptors);
     Assert.assertEquals(1, serviceDescriptors.size());
@@ -125,6 +140,11 @@ public class KerberosDescriptorTest {
     Assert.assertEquals("EXAMPLE.COM", properties.get("realm"));
     Assert.assertEquals("Hello World", properties.get("some.property"));
 
+    Set<String> authToLocalProperties = descriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(1, authToLocalProperties.size());
+    Assert.assertEquals("global.name.rules", authToLocalProperties.iterator().next());
+
     Map<String, KerberosServiceDescriptor> services = descriptor.getServices();
     Assert.assertNotNull(services);
     Assert.assertEquals(1, services.size());
@@ -193,6 +213,14 @@ public class KerberosDescriptorTest {
     Assert.assertEquals("/etc/security/keytabs", properties.get("keytab_dir"));
     Assert.assertEquals("Hello World", properties.get("some.property"));
 
+    Set<String> authToLocalProperties = descriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(2, authToLocalProperties.size());
+    // guarantee ordering...
+    Iterator<String> iterator = new TreeSet<String>(authToLocalProperties).iterator();
+    Assert.assertEquals("generic.name.rules", iterator.next());
+    Assert.assertEquals("global.name.rules", iterator.next());
+
     Map<String, KerberosServiceDescriptor> serviceDescriptors = descriptor.getServices();
     Assert.assertNotNull(serviceDescriptors);
     Assert.assertEquals(2, serviceDescriptors.size());

ambari-server/src/test/java/org/apache/ambari/server/state/kerberos/KerberosServiceDescriptorTest.java

@@ -26,8 +26,11 @@ import java.io.IOException;
 import java.net.URL;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
 
 public class KerberosServiceDescriptorTest {
   public static final String JSON_VALUE =
@@ -39,6 +42,9 @@ public class KerberosServiceDescriptorTest {
           "  \"components\": [" +
           KerberosComponentDescriptorTest.JSON_VALUE +
           "]," +
+          "  \"auth_to_local_properties\": [" +
+          "      service.name.rules1" +
+          "    ]," +
           "  \"configurations\": [" +
           "    {" +
           "      \"service-site\": {" +
@@ -60,6 +66,9 @@ public class KerberosServiceDescriptorTest {
           "  \"components\": [" +
           KerberosComponentDescriptorTest.JSON_VALUE +
           "]," +
+          "  \"auth_to_local_properties\": [" +
+          "      service.name.rules1" +
+          "    ]," +
           "  \"configurations\": [" +
           "    {" +
           "      \"service-site\": {" +
@@ -111,6 +120,9 @@ public class KerberosServiceDescriptorTest {
               }
             });
           }});
+          put(KerberosDescriptorType.AUTH_TO_LOCAL_PROPERTY.getDescriptorPluralName(), new ArrayList<String>() {{
+            add("service.name.rules2");
+          }});
         }
       };
 
@@ -161,6 +173,11 @@ public class KerberosServiceDescriptorTest {
     Assert.assertEquals(2, properties.size());
     Assert.assertEquals("value1", properties.get("service.property1"));
     Assert.assertEquals("value2", properties.get("service.property2"));
+
+    Set<String> authToLocalProperties = serviceDescriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(1, authToLocalProperties.size());
+    Assert.assertEquals("service.name.rules1", authToLocalProperties.iterator().next());
   }
 
   public static void validateFromMap(KerberosServiceDescriptor serviceDescriptor) {
@@ -201,6 +218,11 @@ public class KerberosServiceDescriptorTest {
     Assert.assertEquals(2, properties.size());
     Assert.assertEquals("red", properties.get("service.property1"));
     Assert.assertEquals("green", properties.get("service.property"));
+
+    Set<String> authToLocalProperties = serviceDescriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(1, authToLocalProperties.size());
+    Assert.assertEquals("service.name.rules2", authToLocalProperties.iterator().next());
   }
 
   public void validateUpdatedData(KerberosServiceDescriptor serviceDescriptor) {
@@ -240,6 +262,15 @@ public class KerberosServiceDescriptorTest {
     Assert.assertEquals("red", properties.get("service.property1"));
     Assert.assertEquals("value2", properties.get("service.property2"));
     Assert.assertEquals("green", properties.get("service.property"));
+
+    Set<String> authToLocalProperties = serviceDescriptor.getAuthToLocalProperties();
+    Assert.assertNotNull(authToLocalProperties);
+    Assert.assertEquals(2, authToLocalProperties.size());
+    // guarantee ordering...
+    Iterator<String> iterator = new TreeSet<String>(authToLocalProperties).iterator();
+    Assert.assertEquals("service.name.rules1", iterator.next());
+    Assert.assertEquals("service.name.rules2", iterator.next());
+
   }
 
   private KerberosServiceDescriptor createFromJSON() throws AmbariException {

ambari-server/src/test/resources/stacks/HDP/2.0.8/kerberos.json

@@ -23,13 +23,16 @@
       }
     }
   ],
+  "auth_to_local_properties" : [
+    "hadoop.security.auth_to_local"
+  ],
   "configurations": [
     {
       "core-site": {
         "hadoop.security.authentication": "kerberos",
         "hadoop.rpc.protection": "authentication",
         "hadoop.security.authorization": "true",
-        "hadoop.security.auth_to_local": "_AUTH_TO_LOCAL_RULES",
+        "hadoop.security.auth_to_local": "",
         "hadoop.proxyuser.superuser.hosts": "",
         "hadoop.proxyuser.superuser.groups": ""
       }