AMBARI-9439. Kerberos: Do not validate host health or maintenance state when enabling Kerberos (rlevas)

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

@@ -24,7 +24,6 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -32,6 +31,7 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.regex.Pattern;
 
+import com.google.common.reflect.TypeToken;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.HostNotFoundException;
 import org.apache.ambari.server.RoleCommand;
@@ -51,6 +51,7 @@ import org.apache.ambari.server.events.HostComponentVersionEvent;
 import org.apache.ambari.server.events.publishers.AlertEventPublisher;
 import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
 import org.apache.ambari.server.metadata.ActionMetadata;
+import org.apache.ambari.server.orm.dao.KerberosPrincipalHostDAO;
 import org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFile;
 import org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFileReader;
 import org.apache.ambari.server.serveraction.kerberos.KerberosServerAction;
@@ -82,6 +83,7 @@ import org.apache.ambari.server.state.host.HostHealthyHeartbeatEvent;
 import org.apache.ambari.server.state.host.HostRegistrationRequestEvent;
 import org.apache.ambari.server.state.host.HostStatusUpdatesReceivedEvent;
 import org.apache.ambari.server.state.host.HostUnhealthyHeartbeatEvent;
+import org.apache.ambari.server.state.scheduler.RequestExecution;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpFailedEvent;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpInProgressEvent;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpSucceededEvent;
@@ -91,7 +93,6 @@ import org.apache.ambari.server.utils.StageUtils;
 import org.apache.ambari.server.utils.VersionUtils;
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.codec.digest.DigestUtils;
-import org.apache.commons.csv.CSVParser;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -152,6 +153,12 @@ public class HeartBeatHandler {
   @Inject
   private AmbariEventPublisher ambariEventPublisher;
 
+  /**
+   * KerberosPrincipalHostDAO used to set and get Kerberos principal details
+   */
+  @Inject
+  private KerberosPrincipalHostDAO kerberosPrincipalHostDAO;
+
   private Map<String, Long> hostResponseIds = new ConcurrentHashMap<String, Long>();
 
   private Map<String, HeartBeatResponse> hostResponses = new ConcurrentHashMap<String, HeartBeatResponse>();
@@ -444,6 +451,35 @@ public class HeartBeatHandler {
               report.getStatus().equals("IN_PROGRESS")) {
         hostRoleCommand.setStartTime(now);
       }
+
+      // If the report indicates the keytab file was successfully transferred to a host, record this
+      // for future reference
+      if ("KERBEROS".equalsIgnoreCase(report.getServiceName()) &&
+          "KERBEROS_CLIENT".equalsIgnoreCase(report.getRole()) &&
+          RoleCommand.CUSTOM_COMMAND.name().equalsIgnoreCase(report.getRoleCommand()) &&
+          "SET_KEYTAB".equalsIgnoreCase(report.getCustomCommand()) &&
+          RequestExecution.Status.COMPLETED.name().equalsIgnoreCase(report.getStatus())) {
+
+        Map<String, String> structuredOutput;
+        try {
+          structuredOutput = gson.fromJson(report.getStructuredOut(),
+              new TypeToken<Map<String, String>>() {
+              }.getType());
+        } catch (JsonSyntaxException ex) {
+          //Json structure was incorrect do nothing, pass this data further for processing
+          structuredOutput = 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);
+            }
+          }
+        }
+      }
+
       //pass custom STAR, STOP and RESTART
       if (RoleCommand.ACTIONEXECUTE.toString().equals(report.getRoleCommand()) ||
          (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
@@ -951,54 +987,70 @@ public class HeartBeatHandler {
     return commands;
   }
 
-  static void injectKeytab(ExecutionCommand ec, String targetHost) throws AmbariException {
+  /**
+   * Insert Kerberos keytab details into the ExecutionCommand for the SET_KEYTAB custom command if
+   * any keytab details and associated data exists for the target host.
+   *
+   * @param ec the ExecutionCommand to update
+   * @param targetHost a name of the host the relevant command is destined for
+   * @throws AmbariException
+   */
+  void injectKeytab(ExecutionCommand ec, String targetHost) throws AmbariException {
     Map<String, String> hlp = ec.getHostLevelParams();
     if ((hlp == null) || !"SET_KEYTAB".equals(hlp.get("custom_command"))) {
       return;
     }
     List<Map<String, String>> kcp = ec.getKerberosCommandParams();
     String dataDir = ec.getCommandParams().get(KerberosServerAction.DATA_DIRECTORY);
-    File file = new File(dataDir + File.separator + "index.dat");
-    CSVParser csvParser = null;
+    KerberosActionDataFileReader reader = null;
+
     try {
-      KerberosActionDataFileReader reader = new KerberosActionDataFileReader(file);
-      Iterator<Map<String, String>> iterator = reader.iterator();
-      while (iterator.hasNext())    {
-        Map<String, String> record = iterator.next();
+      reader = new KerberosActionDataFileReader(new File(dataDir, KerberosActionDataFile.DATA_FILE_NAME));
+
+      for(Map<String, String> record : reader) {
         String hostName = record.get(KerberosActionDataFile.HOSTNAME);
-        if (!targetHost.equalsIgnoreCase(hostName))    {
-          continue;
+
+        if (targetHost.equalsIgnoreCase(hostName)) {
+          String keytabFilePath = record.get(KerberosActionDataFile.KEYTAB_FILE_PATH);
+
+          if (keytabFilePath != null) {
+            String sha1Keytab = DigestUtils.sha1Hex(keytabFilePath);
+            File keytabFile = new File(dataDir + File.separator + hostName + File.separator + sha1Keytab);
+
+            if (keytabFile.canRead()) {
+              Map<String, String> keytabMap = new HashMap<String, String>();
+              String principal = record.get(KerberosActionDataFile.PRINCIPAL);
+              String isService = record.get(KerberosActionDataFile.SERVICE);
+
+              keytabMap.put(KerberosActionDataFile.HOSTNAME, hostName);
+              keytabMap.put(KerberosActionDataFile.SERVICE, isService);
+              keytabMap.put(KerberosActionDataFile.COMPONENT, record.get(KerberosActionDataFile.COMPONENT));
+              keytabMap.put(KerberosActionDataFile.PRINCIPAL, principal);
+              keytabMap.put(KerberosActionDataFile.PRINCIPAL_CONFIGURATION, record.get(KerberosActionDataFile.PRINCIPAL_CONFIGURATION));
+              keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_PATH, keytabFilePath);
+              keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME, record.get(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME));
+              keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS, record.get(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS));
+              keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME, record.get(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME));
+              keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS, record.get(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS));
+              keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION, record.get(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION));
+
+              BufferedInputStream bufferedIn = new BufferedInputStream(new FileInputStream(keytabFile));
+              byte[] keytabContent = IOUtils.toByteArray(bufferedIn);
+              String keytabContentBase64 = Base64.encodeBase64String(keytabContent);
+              keytabMap.put(KerberosServerAction.KEYTAB_CONTENT_BASE64, keytabContentBase64);
+
+              kcp.add(keytabMap);
+            }
+          }
         }
-        Map<String, String> keytabMap = new HashMap<String, String>();
-        keytabMap.put(KerberosActionDataFile.HOSTNAME, hostName);
-        keytabMap.put(KerberosActionDataFile.SERVICE, record.get(KerberosActionDataFile.SERVICE));
-        keytabMap.put(KerberosActionDataFile.COMPONENT, record.get(KerberosActionDataFile.COMPONENT));
-        keytabMap.put(KerberosActionDataFile.PRINCIPAL, record.get(KerberosActionDataFile.PRINCIPAL));
-        keytabMap.put(KerberosActionDataFile.PRINCIPAL_CONFIGURATION, record.get(KerberosActionDataFile.PRINCIPAL_CONFIGURATION));
-        keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_PATH, record.get(KerberosActionDataFile.KEYTAB_FILE_PATH));
-        keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME, record.get(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME));
-        keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS, record.get(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS));
-        keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME, record.get(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME));
-        keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS, record.get(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS));
-        keytabMap.put(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION, record.get(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION));
-
-        String sha1Keytab =  DigestUtils.sha1Hex(record.get(KerberosActionDataFile.KEYTAB_FILE_PATH));
-
-        BufferedInputStream bufferedIn = new BufferedInputStream(
-          new FileInputStream(dataDir + File.separator +
-            hostName + File.separator + sha1Keytab));
-        byte[] keytabContent = IOUtils.toByteArray(bufferedIn);
-        String keytabContentBase64 = Base64.encodeBase64String(keytabContent);
-        keytabMap.put(KerberosServerAction.KEYTAB_CONTENT_BASE64, keytabContentBase64);
-        kcp.add(keytabMap);
       }
     } catch (IOException e) {
       throw new AmbariException("Could not inject keytabs to enable kerberos");
     }  finally {
-      if (csvParser != null && !csvParser.isClosed())  {
+      if (reader != null) {
         try {
-          csvParser.close();
-        }  catch (Throwable t)  {
+          reader.close();
+        } catch (Throwable t) {
           // ignored
         }
       }

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

@@ -48,6 +48,7 @@ import org.apache.ambari.server.metadata.RoleCommandOrder;
 import org.apache.ambari.server.serveraction.ServerAction;
 import org.apache.ambari.server.serveraction.kerberos.CreateKeytabFilesServerAction;
 import org.apache.ambari.server.serveraction.kerberos.CreatePrincipalsServerAction;
+import org.apache.ambari.server.serveraction.kerberos.DestroyPrincipalsServerAction;
 import org.apache.ambari.server.serveraction.kerberos.FinalizeKerberosServerAction;
 import org.apache.ambari.server.serveraction.kerberos.KDCType;
 import org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFile;
@@ -71,13 +72,11 @@ import org.apache.ambari.server.state.Config;
 import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
-import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.PropertyInfo;
 import org.apache.ambari.server.state.SecurityState;
 import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.Service;
 import org.apache.ambari.server.state.ServiceComponentHost;
-import org.apache.ambari.server.state.ServiceInfo;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.kerberos.KerberosComponentDescriptor;
 import org.apache.ambari.server.state.kerberos.KerberosConfigurationDescriptor;
@@ -123,9 +122,6 @@ public class KerberosHelper {
   @Inject
   private AmbariCustomCommandExecutionHelper customCommandExecutionHelper;
 
-  @Inject
-  private MaintenanceStateHelper maintenanceStateHelper;
-
   @Inject
   private AmbariManagementController ambariManagementController;
 
@@ -238,9 +234,14 @@ public class KerberosHelper {
                 throw new AmbariException(String.format("Custom operation %s can only be requested with the security type cluster property: %s", operation.name(), SecurityType.KERBEROS.name()));
               }
 
-              if ("true".equalsIgnoreCase(value)) {
-                requestStageContainer = handle(cluster, getKerberosDetails(cluster), null, null, requestStageContainer, new CreatePrincipalsAndKeytabsHandler());
+              if ("true".equalsIgnoreCase(value) || "all".equalsIgnoreCase(value)) {
+                requestStageContainer = handle(cluster, getKerberosDetails(cluster), null, null, requestStageContainer, new CreatePrincipalsAndKeytabsHandler(true));
+              } else if ("missing".equalsIgnoreCase(value)) {
+                requestStageContainer = handle(cluster, getKerberosDetails(cluster), null, null, requestStageContainer, new CreatePrincipalsAndKeytabsHandler(false));
+              } else {
+                throw new AmbariException(String.format("Unexpected directive value: %s", value));
               }
+
               break;
 
             default: // No other operations are currently supported
@@ -285,7 +286,7 @@ public class KerberosHelper {
                                                 Collection<String> identityFilter, RequestStageContainer requestStageContainer)
       throws AmbariException {
     return handle(cluster, getKerberosDetails(cluster), serviceComponentFilter, identityFilter,
-        requestStageContainer, new CreatePrincipalsAndKeytabsHandler());
+        requestStageContainer, new CreatePrincipalsAndKeytabsHandler(false));
   }
 
   /**
@@ -356,98 +357,94 @@ public class KerberosHelper {
           // component (aka service component host - sch) determine the configuration updates and
           // and the principals an keytabs to create.
           for (Host host : hosts.values()) {
-            // Only process "healthy" hosts.  When an unhealthy host becomes healthy, it will be
-            // processed the next time this executes.
-            if (host.getState() == HostState.HEALTHY) {
-              String hostname = host.getHostName();
-
-              // Get a list of components on the current host
-              List<ServiceComponentHost> serviceComponentHosts = cluster.getServiceComponentHosts(hostname);
-
-              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);
-
-                // 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>>();
-                }
+            String hostname = host.getHostName();
 
-                // 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);
-                }
+            // Get a list of components on the current host
+            List<ServiceComponentHost> serviceComponentHosts = cluster.getServiceComponentHosts(hostname);
 
-                // Add the current hostname under "host" and "hostname"
-                generalProperties.put("host", hostname);
-                generalProperties.put("hostname", hostname);
-                generalProperties.put("cluster_name", clusterName);
+            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);
 
-                if (configurations.get("") == null) {
-                  configurations.put("", generalProperties);
-                } else {
-                  configurations.get("").putAll(generalProperties);
-                }
+              // 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);
+              }
 
-                // Iterate over the components installed on the current host to get the service and
-                // component-level Kerberos descriptors in order to determine which principals,
-                // keytab files, and configurations need to be created or updated.
-                for (ServiceComponentHost sch : serviceComponentHosts) {
-                  String serviceName = sch.getServiceName();
-
-                  // If there is no filter or the filter contains the current service name...
-                  if ((serviceComponentFilter == null) || serviceComponentFilter.containsKey(serviceName)) {
-                    Collection<String> componentFilter = (serviceComponentFilter == null) ? null : serviceComponentFilter.get(serviceName);
-                    KerberosServiceDescriptor serviceDescriptor = kerberosDescriptor.getService(serviceName);
-
-                    if (serviceDescriptor != null) {
-                      String componentName = sch.getServiceComponentName();
-
-                      // 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);
+              // Iterate over the components installed on the current host to get the service and
+              // component-level Kerberos descriptors in order to determine which principals,
+              // keytab files, and configurations need to be created or updated.
+              for (ServiceComponentHost sch : serviceComponentHosts) {
+                String serviceName = sch.getServiceName();
+
+                // If there is no filter or the filter contains the current service name...
+                if ((serviceComponentFilter == null) || serviceComponentFilter.containsKey(serviceName)) {
+                  Collection<String> componentFilter = (serviceComponentFilter == null) ? null : serviceComponentFilter.get(serviceName);
+                  KerberosServiceDescriptor serviceDescriptor = kerberosDescriptor.getService(serviceName);
+
+                  if (serviceDescriptor != null) {
+                    String componentName = sch.getServiceComponentName();
+
+                    // 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 to auth_to_local builder
-                        addIdentities(authToLocalBuilder, serviceIdentities, identityFilter, configurations);
+                        // 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);
                     }
                   }
                 }
@@ -1204,15 +1201,33 @@ public class KerberosHelper {
    * Given a Collection of ServiceComponentHosts generates a unique list of hosts.
    *
    * @param serviceComponentHosts a Collection of ServiceComponentHosts from which to to retrieve host names
+   * @param allowedStates         a Set of HostStates to use to filter the list of hosts, if null, no filter is applied
    * @return a List of (unique) host names
+   * @throws org.apache.ambari.server.AmbariException
    */
-  private List<String> createUniqueHostList(Collection<ServiceComponentHost> serviceComponentHosts) {
+  private List<String> createUniqueHostList(Collection<ServiceComponentHost> serviceComponentHosts, Set<HostState> allowedStates)
+      throws AmbariException {
     Set<String> hostNames = new HashSet<String>();
+    Set<String> visitedHostNames = new HashSet<String>();
 
     if (serviceComponentHosts != null) {
-
       for (ServiceComponentHost sch : serviceComponentHosts) {
-        hostNames.add(sch.getHostName());
+        String hostname = sch.getHostName();
+        if (!visitedHostNames.contains(hostname)) {
+          // If allowedStates is null, assume the caller doesnt care about the state of the host
+          // so skip the call to get the relevant Host data and just add the host to the list
+          if (allowedStates == null) {
+            hostNames.add(hostname);
+          } else {
+            Host host = clusters.getHost(hostname);
+
+            if (allowedStates.contains(host.getState())) {
+              hostNames.add(hostname);
+            }
+          }
+
+          visitedHostNames.add(hostname);
+        }
       }
     }
 
@@ -1336,6 +1351,29 @@ public class KerberosHelper {
       requestStageContainer.addStages(roleGraph.getStages());
     }
 
+    public void addDestroyPrincipalsStage(Cluster cluster, String clusterHostInfoJson,
+                                          String hostParamsJson, ServiceComponentHostServerActionEvent event,
+                                          Map<String, String> commandParameters,
+                                          RoleCommandOrder roleCommandOrder, RequestStageContainer requestStageContainer)
+        throws AmbariException {
+      Stage stage = createServerActionStage(requestStageContainer.getLastStageId(),
+          cluster,
+          requestStageContainer.getId(),
+          "Destroy Principals",
+          clusterHostInfoJson,
+          "{}",
+          hostParamsJson,
+          DestroyPrincipalsServerAction.class,
+          event,
+          commandParameters,
+          "Destroy Principals",
+          1200);
+
+      RoleGraph roleGraph = new RoleGraph(roleCommandOrder);
+      roleGraph.build(stage);
+      requestStageContainer.addStages(roleGraph.getStages());
+    }
+
     public void addCreateKeytabFilesStage(Cluster cluster, String clusterHostInfoJson,
                                           String hostParamsJson, ServiceComponentHostServerActionEvent event,
                                           Map<String, String> commandParameters,
@@ -1374,7 +1412,7 @@ public class KerberosHelper {
           hostParamsJson);
 
       if (!serviceComponentHosts.isEmpty()) {
-        List<String> hostsToUpdate = createUniqueHostList(serviceComponentHosts);
+        List<String> hostsToUpdate = createUniqueHostList(serviceComponentHosts, Collections.singleton(HostState.HEALTHY));
         Map<String, String> requestParams = new HashMap<String, String>();
         List<RequestResourceFilter> requestResourceFilters = new ArrayList<RequestResourceFilter>();
         RequestResourceFilter reqResFilter = new RequestResourceFilter("KERBEROS", "KERBEROS_CLIENT", hostsToUpdate);
@@ -1437,7 +1475,6 @@ public class KerberosHelper {
     @Override
     public boolean shouldProcess(SecurityState desiredSecurityState, ServiceComponentHost sch) throws AmbariException {
       return (desiredSecurityState == SecurityState.SECURED_KERBEROS) &&
-          (maintenanceStateHelper.getEffectiveState(sch) == MaintenanceState.OFF) &&
           (sch.getSecurityState() != SecurityState.SECURED_KERBEROS) &&
           (sch.getSecurityState() != SecurityState.SECURING);
     }
@@ -1567,7 +1604,6 @@ public class KerberosHelper {
     @Override
     public boolean shouldProcess(SecurityState desiredSecurityState, ServiceComponentHost sch) throws AmbariException {
       return (desiredSecurityState == SecurityState.UNSECURED) &&
-          (maintenanceStateHelper.getEffectiveState(sch) == MaintenanceState.OFF) &&
           ((sch.getDesiredSecurityState() != SecurityState.UNSECURED) || (sch.getSecurityState() != SecurityState.UNSECURED)) &&
           (sch.getSecurityState() != SecurityState.UNSECURING);
     }
@@ -1685,6 +1721,11 @@ public class KerberosHelper {
       addUpdateConfigurationsStage(cluster, clusterHostInfoJson, hostParamsJson, event, commandParameters,
           roleCommandOrder, requestStageContainer);
 
+      // *****************************************************************
+      // Create stage to remove principals
+      addDestroyPrincipalsStage(cluster, clusterHostInfoJson, hostParamsJson, event, commandParameters,
+          roleCommandOrder, requestStageContainer);
+
       return requestStageContainer.getLastStageId();
     }
   }
@@ -1702,9 +1743,27 @@ public class KerberosHelper {
    * </ol>
    */
   private class CreatePrincipalsAndKeytabsHandler extends Handler {
+    /**
+     * A boolean value indicating whether to create keytabs for all principals (<code>true</code>)
+     * or only the ones that are missing (<code>false</code>).
+     */
+    private boolean regenerateAllKeytabs;
+
+    /**
+     * CreatePrincipalsAndKeytabsHandler constructor to set whether this instance should be used to
+     * regenerate all keytabs or just the ones that have not been distributed
+     *
+     * @param regenerateAllKeytabs A boolean value indicating whether to create keytabs for all
+     *                             principals (<code>true</code> or only the ones that are missing
+     *                             (<code>false</code>)
+     */
+    public CreatePrincipalsAndKeytabsHandler(boolean regenerateAllKeytabs) {
+      this.regenerateAllKeytabs = regenerateAllKeytabs;
+    }
+
     @Override
     public boolean shouldProcess(SecurityState desiredSecurityState, ServiceComponentHost sch) throws AmbariException {
-      return (maintenanceStateHelper.getEffectiveState(sch) == MaintenanceState.OFF);
+      return true;
     }
 
     @Override
@@ -1750,10 +1809,11 @@ public class KerberosHelper {
       commandParameters.put(KerberosServerAction.DEFAULT_REALM, kerberosDetails.getDefaultRealm());
       commandParameters.put(KerberosServerAction.KDC_TYPE, kerberosDetails.getKdcType().name());
       commandParameters.put(KerberosServerAction.ADMINISTRATOR_CREDENTIAL, getEncryptedAdministratorCredentials(cluster));
+      commandParameters.put(KerberosServerAction.REGENERATE_ALL, (regenerateAllKeytabs) ? "true" : "false");
 
       // *****************************************************************
       // Create stage to create principals
-      super.addCreatePrincipalsStage(cluster, clusterHostInfoJson, hostParamsJson, event,
+      addCreatePrincipalsStage(cluster, clusterHostInfoJson, hostParamsJson, event,
           commandParameters, roleCommandOrder, requestStageContainer);
 
       // *****************************************************************
@@ -1775,8 +1835,8 @@ public class KerberosHelper {
    * It is required that the SecurityType from the request is wither KERBEROS or NONE and that at least one
    * directive in the requestProperties map is supported.
    *
-   * @param requestSecurityType      the SecurityType from the request
-   * @param requestProperties A Map of request directives and their values
+   * @param requestSecurityType the SecurityType from the request
+   * @param requestProperties   A Map of request directives and their values
    * @return true if custom operations should be executed; false otherwise
    */
   public boolean shouldExecuteCustomOperations(SecurityType requestSecurityType, Map<String, String> requestProperties) {

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

@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.orm.dao;
+
+
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.google.inject.Singleton;
+import com.google.inject.persist.Transactional;
+import org.apache.ambari.server.orm.RequiresSession;
+import org.apache.ambari.server.orm.entities.KerberosPrincipalEntity;
+
+import javax.persistence.EntityManager;
+import javax.persistence.TypedQuery;
+import java.util.List;
+
+
+/**
+ * HostKerberosPrincipal Data Access Object.
+ */
+@Singleton
+public class KerberosPrincipalDAO {
+
+  /**
+   * JPA entity manager
+   */
+  @Inject
+  Provider<EntityManager> entityManagerProvider;
+
+
+  /**
+   * Make an instance managed and persistent.
+   *
+   * @param kerberosPrincipalEntity entity to persist
+   */
+  @Transactional
+  public void create(KerberosPrincipalEntity kerberosPrincipalEntity) {
+    entityManagerProvider.get().persist(kerberosPrincipalEntity);
+  }
+
+  /**
+   * Make an instance managed and persistent.
+   *
+   * @param principalName the principal name to use when creating a new KerberosPrincipalEntity to
+   *                      store
+   * @param service       a boolean value declaring whether the principal represents a service (true) or not )false).
+   */
+  @Transactional
+  public void create(String principalName, boolean service) {
+    create(new KerberosPrincipalEntity(principalName, service, null));
+  }
+
+  /**
+   * Merge the state of the given entity into the current persistence context.
+   *
+   * @param kerberosPrincipalEntity entity to merge
+   * @return the merged entity
+   */
+  @Transactional
+  public KerberosPrincipalEntity merge(KerberosPrincipalEntity kerberosPrincipalEntity) {
+    return entityManagerProvider.get().merge(kerberosPrincipalEntity);
+  }
+
+  /**
+   * Remove the entity instance.
+   *
+   * @param kerberosPrincipalEntity entity to remove
+   */
+  @Transactional
+  public void remove(KerberosPrincipalEntity kerberosPrincipalEntity) {
+    entityManagerProvider.get().remove(merge(kerberosPrincipalEntity));
+  }
+
+  /**
+   * Remove entity instance by primary key
+   *
+   * @param principalName Primary key: kerberos principal name
+   */
+  @Transactional
+  public void remove(String principalName) {
+    entityManagerProvider.get().remove(find(principalName));
+  }
+
+  /**
+   * Refresh the state of the instance from the database,
+   * overwriting changes made to the entity, if any.
+   *
+   * @param kerberosPrincipalEntity entity to refresh
+   */
+  @Transactional
+  public void refresh(KerberosPrincipalEntity kerberosPrincipalEntity) {
+    entityManagerProvider.get().refresh(kerberosPrincipalEntity);
+  }
+
+
+  /**
+   * Find a KerberosPrincipalEntity with the given principal name.
+   *
+   * @param principalName name of kerberos principal to find
+   * @return a matching KerberosPrincipalEntity or null
+   */
+  @RequiresSession
+  public KerberosPrincipalEntity find(String principalName) {
+    return entityManagerProvider.get().find(KerberosPrincipalEntity.class, principalName);
+  }
+
+  /**
+   * Find all kerberos principals.
+   *
+   * @return a List of all KerberosPrincipalEntity objects or an empty List
+   */
+  @RequiresSession
+  public List<KerberosPrincipalEntity> findAll() {
+    TypedQuery<KerberosPrincipalEntity> query = entityManagerProvider.get()
+        .createNamedQuery("KerberosPrincipalEntityFindAll", KerberosPrincipalEntity.class);
+    return query.getResultList();
+  }
+
+  /**
+   * Tests the given principal name to see if it already exists.
+   *
+   * @param principalName name of kerberos principal to find
+   * @return true if the principal exists; false otherwise
+   */
+  @RequiresSession
+  public boolean exists(String principalName) {
+    return find(principalName) != null;
+  }
+
+}

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

@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.orm.dao;
+
+
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.google.inject.Singleton;
+import com.google.inject.persist.Transactional;
+import org.apache.ambari.server.orm.RequiresSession;
+import org.apache.ambari.server.orm.entities.KerberosPrincipalHostEntity;
+import org.apache.ambari.server.orm.entities.KerberosPrincipalHostEntityPK;
+
+import javax.persistence.EntityManager;
+import javax.persistence.TypedQuery;
+import java.util.List;
+
+
+/**
+ * HostKerberosPrincipal Data Access Object.
+ */
+@Singleton
+public class KerberosPrincipalHostDAO {
+
+  /**
+   * JPA entity manager
+   */
+  @Inject
+  Provider<EntityManager> entityManagerProvider;
+
+  /**
+   * Make an instance managed and persistent.
+   *
+   * @param kerberosPrincipalHostEntity entity to persist
+   */
+  @Transactional
+  public void create(KerberosPrincipalHostEntity kerberosPrincipalHostEntity) {
+    entityManagerProvider.get().persist(kerberosPrincipalHostEntity);
+  }
+
+  public void create(String principal, String hostName) {
+    create(new KerberosPrincipalHostEntity(principal, hostName));
+  }
+
+  /**
+   * Merge the state of the given entity into the current persistence context.
+   *
+   * @param kerberosPrincipalHostEntity entity to merge
+   * @return the merged entity
+   */
+  @Transactional
+  public KerberosPrincipalHostEntity merge(KerberosPrincipalHostEntity kerberosPrincipalHostEntity) {
+    return entityManagerProvider.get().merge(kerberosPrincipalHostEntity);
+  }
+
+  /**
+   * Remove the entity instance.
+   *
+   * @param kerberosPrincipalHostEntity entity to remove
+   */
+  @Transactional
+  public void remove(KerberosPrincipalHostEntity kerberosPrincipalHostEntity) {
+    entityManagerProvider.get().remove(merge(kerberosPrincipalHostEntity));
+  }
+
+
+  /**
+   * Refresh the state of the instance from the database,
+   * overwriting changes made to the entity, if any.
+   *
+   * @param kerberosPrincipalHostEntity entity to refresh
+   */
+  @Transactional
+  public void refresh(KerberosPrincipalHostEntity kerberosPrincipalHostEntity) {
+    entityManagerProvider.get().refresh(kerberosPrincipalHostEntity);
+  }
+
+
+  /**
+   * Finds KerberosPrincipalHostEntities for the requested principal
+   *
+   * @param principalName a String indicating the name of the requested principal
+   * @return a List of requested KerberosPrincipalHostEntities or null if none were found
+   */
+  @RequiresSession
+  public List<KerberosPrincipalHostEntity> findByPrincipal(String principalName) {
+    final TypedQuery<KerberosPrincipalHostEntity> query = entityManagerProvider.get()
+        .createNamedQuery("KerberosPrincipalHostEntityFindByPrincipal", KerberosPrincipalHostEntity.class);
+    query.setParameter("principalName", principalName);
+    return query.getResultList();
+  }
+
+  /**
+   * Find KerberosPrincipalHostEntities for the requested host
+   *
+   * @param hostName a String indicating the name of the requested host
+   * @return a List of requested KerberosPrincipalHostEntities or null if none were found
+   */
+  @RequiresSession
+  public List<KerberosPrincipalHostEntity> findByHost(String hostName) {
+    final TypedQuery<KerberosPrincipalHostEntity> query = entityManagerProvider.get()
+        .createNamedQuery("KerberosPrincipalHostEntityFindByHost", KerberosPrincipalHostEntity.class);
+    query.setParameter("hostName", hostName);
+    return query.getResultList();
+  }
+
+  /**
+   * Find the KerberosPrincipalHostEntity for the specified primary key
+   *
+   * @param primaryKey a KerberosPrincipalHostEntityPK containing the requested principal and host names
+   * @return the KerberosPrincipalHostEntity or null if not found
+   */
+  @RequiresSession
+  public KerberosPrincipalHostEntity find(KerberosPrincipalHostEntityPK primaryKey) {
+    return entityManagerProvider.get().find(KerberosPrincipalHostEntity.class, primaryKey);
+  }
+
+  /**
+   * Find the KerberosPrincipalHostEntity for the requested principal name and host
+   *
+   * @param principalName a String indicating the name of the requested principal
+   * @param hostName      a String indicating the name of the requested host
+   * @return the KerberosPrincipalHostEntity or null if not found
+   */
+  @RequiresSession
+  public KerberosPrincipalHostEntity find(String principalName, String hostName) {
+    return entityManagerProvider.get().find(KerberosPrincipalHostEntity.class,
+        new KerberosPrincipalHostEntityPK(principalName, hostName));
+  }
+
+  /**
+   * Find all KerberosPrincipalHostEntities.
+   *
+   * @return a List of requested KerberosPrincipalHostEntities or null if none were found
+   */
+  @RequiresSession
+  public List<KerberosPrincipalHostEntity> findAll() {
+    TypedQuery<KerberosPrincipalHostEntity> query = entityManagerProvider.get().
+        createNamedQuery("KerberosPrincipalHostEntityFindAll", KerberosPrincipalHostEntity.class);
+
+    return query.getResultList();
+  }
+
+
+  /**
+   * Remove KerberosPrincipalHostEntity instances for the specified principal name
+   *
+   * @param principalName a String indicating the name of the principal
+   */
+  @Transactional
+  public void removeByPrincipal(String principalName) {
+    entityManagerProvider.get().remove(findByPrincipal(principalName));
+  }
+
+  /**
+   * Remove KerberosPrincipalHostEntity instances for the specified host
+   *
+   * @param hostName a String indicating the name of the host
+   */
+  @Transactional
+  public void removeByHost(String hostName) {
+    entityManagerProvider.get().remove(findByHost(hostName));
+  }
+
+  /**
+   * Tests the existence of a principal on at least one host
+   *
+   * @param principalName a String indicating the name of the principal to test
+   * @return true if a principal is related to one or more hosts; otherwise false
+   */
+  @RequiresSession
+  public boolean exists(String principalName) {
+    List<KerberosPrincipalHostEntity> foundEntries = findByPrincipal(principalName);
+    return (foundEntries != null) && !foundEntries.isEmpty();
+  }
+
+  /**
+   * Tests the existence of a particular principal on a specific host
+   *
+   * @param principalName a String indicating the name of the principal to test
+   * @param hostName      a String indicating the name of the host to test
+   * @return true if the requested principal exists
+   */
+  @RequiresSession
+  public boolean exists(String principalName, String hostName) {
+    return find(principalName, hostName) != null;
+  }
+}

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

@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.orm.entities;
+
+import org.apache.ambari.server.orm.dao.KerberosPrincipalHostDAO;
+
+import javax.persistence.CascadeType;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.FetchType;
+import javax.persistence.Id;
+import javax.persistence.JoinColumn;
+import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
+import javax.persistence.OneToMany;
+import javax.persistence.Table;
+import java.util.Collection;
+
+/**
+ * Entity representing a KerberosPrincipal.
+ * <p/>
+ * Each KerberosPrincipal is related to a host, therefore there may be several entities with the same
+ * principal name, but related to different hosts.
+ */
+@Entity
+@Table(name = "kerberos_principal")
+@NamedQueries({
+    @NamedQuery(name = "KerberosPrincipalEntityFindAll",
+        query = "SELECT kp FROM KerberosPrincipalEntity kp")
+})
+public class KerberosPrincipalEntity {
+
+  @Id
+  @Column(name = "principal_name", insertable = true, updatable = false, nullable = false)
+  private String principalName = null;
+
+  @Column(name = "is_service", insertable = true, updatable = false, nullable = false)
+  private Integer service = 1;
+
+  @Column(name = "cached_keytab_path", insertable = true, updatable = true, nullable = true)
+  private String cachedKeytabPath = null;
+
+  @OneToMany(mappedBy = "principalEntity", cascade = CascadeType.REMOVE, fetch = FetchType.LAZY)
+  private Collection<KerberosPrincipalHostEntity> kerberosPrincipalHostEntities;
+
+  /**
+   * Constructs an empty KerberosPrincipalEntity
+   */
+  public KerberosPrincipalEntity() {
+  }
+
+  /**
+   * Constructs a new KerberosPrincipalEntity
+   *
+   * @param principalName    a String declaring the principal name
+   * @param service          a boolean value indicating whether the principal is a service principal (true) or not (false)
+   * @param cachedKeytabPath a String declaring the location of the related cached keytab
+   */
+  public KerberosPrincipalEntity(String principalName, boolean service, String cachedKeytabPath) {
+    setPrincipalName(principalName);
+    setService(service);
+    setCachedKeytabPath(cachedKeytabPath);
+  }
+
+  /**
+   * Gets the principal name for this KerberosPrincipalEntity
+   *
+   * @return a String indicating this KerberosPrincipalEntity's principal name
+   */
+  public String getPrincipalName() {
+    return principalName;
+  }
+
+  /**
+   * Sets the principal name for this KerberosPrincipalEntity
+   *
+   * @param principalName a String indicating this KerberosPrincipalEntity's principal name
+   */
+  public void setPrincipalName(String principalName) {
+    this.principalName = principalName;
+  }
+
+  /**
+   * Indicates whether this KerberosPrincipalEntity represents a service principal (true) or not (false)
+   *
+   * @return true if this KerberosPrincipalEntity represents a service principal; otherwise false
+   */
+  public boolean isService() {
+    return (service == 1);
+  }
+
+  /**
+   * Set whether this KerberosPrincipalEntity represents a service principal (true) or not (false)
+   *
+   * @param service true if this KerberosPrincipalEntity represents a service principal; otherwise false
+   */
+  public void setService(boolean service) {
+    this.service = (service) ? 1 : 0;
+  }
+
+  /**
+   * Gets the location of the cached keytab file, if one exists
+   *
+   * @return a String the location of the cached keytab file, or not if one does not exist
+   */
+  public String getCachedKeytabPath() {
+    return cachedKeytabPath;
+  }
+
+  /**
+   * Sets the location of the cached keytab file, if one exists
+   *
+   * @param cachedKeytabPath a String the location of the cached keytab file, or not if one does not exist
+   */
+  public void setCachedKeytabPath(String cachedKeytabPath) {
+    this.cachedKeytabPath = cachedKeytabPath;
+  }
+
+  /**
+   * Gets the list of related KerberosPrincipalHostEntities
+   *
+   * @return a List of related KerberosPrincipalHostEntities or null if none exist
+   */
+  public Collection<KerberosPrincipalHostEntity> getKerberosPrincipalHostEntities() {
+    return kerberosPrincipalHostEntities;
+  }
+
+  /**
+   * Sets the list of related KerberosPrincipalHostEntities
+   *
+   * @param kerberosPrincipalHostEntities a List of related KerberosPrincipalHostEntities or null if none exist
+   */
+  public void setKerberosPrincipalHostEntities(Collection<KerberosPrincipalHostEntity> kerberosPrincipalHostEntities) {
+    this.kerberosPrincipalHostEntities = kerberosPrincipalHostEntities;
+  }
+}

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

@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.orm.entities;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.IdClass;
+import javax.persistence.JoinColumn;
+import javax.persistence.JoinColumns;
+import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
+import javax.persistence.Table;
+
+/**
+ * Entity representing a KerberosPrincipal stored on a host.
+ */
+@Entity
+@IdClass(KerberosPrincipalHostEntityPK.class)
+@Table(name = "kerberos_principal_host")
+@NamedQueries({
+    @NamedQuery(name = "KerberosPrincipalHostEntityFindAll",
+        query = "SELECT kph FROM KerberosPrincipalHostEntity kph"),
+    @NamedQuery(name = "KerberosPrincipalHostEntityFindByPrincipal",
+        query = "SELECT kph FROM KerberosPrincipalHostEntity kph WHERE kph.principalName=:principalName"),
+    @NamedQuery(name = "KerberosPrincipalHostEntityFindByHost",
+        query = "SELECT kph FROM KerberosPrincipalHostEntity kph WHERE kph.hostName=:hostName")
+})
+public class KerberosPrincipalHostEntity {
+
+  @Id
+  @Column(name = "principal_name", insertable = true, updatable = false, nullable = false)
+  private String principalName;
+
+  @Id
+  @Column(name = "host_name", insertable = true, updatable = false, nullable = false)
+  private String hostName;
+
+  @ManyToOne
+  @JoinColumn(name = "principal_name", referencedColumnName = "principal_name", nullable = false, insertable = false, updatable = false)
+  private KerberosPrincipalEntity principalEntity;
+
+  @ManyToOne
+  @JoinColumn(name = "host_name", referencedColumnName = "host_name", nullable = false, insertable = false, updatable = false)
+  private HostEntity hostEntity;
+
+  /**
+   * Constucts an empty KerberosPrincipalHostEntity
+   */
+  public KerberosPrincipalHostEntity() {
+  }
+
+  /**
+   * Constructs a new KerberosPrincipalHostEntity
+   *
+   * @param principalName a String indicating this KerberosPrincipalHostEntity's principal name
+   * @param hostName      a String indicating the KerberosPrincipalHostEntity's host name
+   */
+  public KerberosPrincipalHostEntity(String principalName, String hostName) {
+    setPrincipalName(principalName);
+    setHostName(hostName);
+  }
+
+  /**
+   * Gets the principal name for this KerberosPrincipalHostEntity
+   *
+   * @return a String indicating this KerberosPrincipalHostEntity's principal name
+   */
+  public String getPrincipalName() {
+    return principalName;
+  }
+
+  /**
+   * Sets the principal name for this KerberosPrincipalHostEntity
+   *
+   * @param principalName a String indicating this KerberosPrincipalHostEntity's principal name
+   */
+  public void setPrincipalName(String principalName) {
+    this.principalName = principalName;
+  }
+
+  /**
+   * Gets the host name for this KerberosHostHostEntity
+   *
+   * @return a String indicating this KerberosHostHostEntity's host name
+   */
+  public String getHostName() {
+    return hostName;
+  }
+
+  /**
+   * Sets the host name for this KerberosHostHostEntity
+   *
+   * @param hostName a String indicating this KerberosHostHostEntity's host name
+   */
+  public void setHostName(String hostName) {
+    this.hostName = hostName;
+  }
+
+  /**
+   * Gets the related HostEntity
+   *
+   * @return the related HostEntity
+   */
+  public HostEntity getHostEntity() {
+    return hostEntity;
+  }
+
+  /**
+   * Sets the related HostEntity
+   *
+   * @param hostEntity the related HostEntity
+   */
+  public void setHostEntity(HostEntity hostEntity) {
+    this.hostEntity = hostEntity;
+  }
+
+  /**
+   * Gets the related KerberosPrincipalEntity
+   *
+   * @return the related KerberosPrincipalEntity
+   */
+  public KerberosPrincipalEntity getPrincipalEntity() {
+    return principalEntity;
+  }
+
+  /**
+   * Sets the related KerberosPrincipalEntity
+   *
+   * @param principalEntity the related KerberosPrincipalEntity
+   */
+  public void setPrincipalEntity(KerberosPrincipalEntity principalEntity) {
+    this.principalEntity = principalEntity;
+  }
+}

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

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.orm.entities;
+
+import javax.persistence.Column;
+import javax.persistence.Id;
+import java.io.Serializable;
+
+/**
+ * Composite primary key for KerberosPrincipalHostEntity.
+ */
+public class KerberosPrincipalHostEntityPK implements Serializable{
+
+  @Id
+  @Column(name = "principal_name", insertable = false, updatable = false, nullable = false)
+  private String principalName = null;
+
+  @Id
+  @Column(name = "host_name", insertable = false, updatable = false, nullable = false)
+  private String hostName = null;
+
+  public KerberosPrincipalHostEntityPK() {
+  }
+
+  public KerberosPrincipalHostEntityPK(String principalName, String hostName) {
+    setPrincipalName(principalName);
+    setHostName(hostName);
+  }
+
+  /**
+   * Get the name of the associated principal.
+   *
+   * @return principal name
+   */
+  public String getPrincipalName() {
+    return principalName;
+  }
+
+  /**
+   * Set the name of the associated principal.
+   *
+   * @param principalName principal name
+   */
+  public void setPrincipalName(String principalName) {
+    this.principalName = principalName;
+  }
+
+  /**
+   * Get the host name.
+   *
+   * @return host name
+   */
+  public String getHostName() {
+    return hostName;
+  }
+
+  /**
+   * Set the configuration type.
+   *
+   * @param hostName host name
+   */
+  public void setHostName(String hostName) {
+    this.hostName = hostName;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    KerberosPrincipalHostEntityPK that = (KerberosPrincipalHostEntityPK) o;
+
+    return this.principalName.equals(that.principalName) &&
+        this.hostName.equals(that.hostName);
+  }
+
+  @Override
+  public int hashCode() {
+    return 31 * principalName.hashCode() + hostName.hashCode();
+  }
+}

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

@@ -18,14 +18,20 @@
 
 package org.apache.ambari.server.serveraction.kerberos;
 
+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.orm.dao.KerberosPrincipalDAO;
+import org.apache.ambari.server.orm.dao.KerberosPrincipalHostDAO;
+import org.apache.ambari.server.orm.entities.KerberosPrincipalEntity;
 import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 
@@ -45,6 +51,18 @@ import static org.apache.ambari.server.serveraction.kerberos.KerberosActionDataF
 public class CreateKeytabFilesServerAction extends KerberosServerAction {
   private final static Logger LOG = LoggerFactory.getLogger(CreateKeytabFilesServerAction.class);
 
+  /**
+   * KerberosPrincipalDAO used to set and get Kerberos principal details
+   */
+  @Inject
+  private KerberosPrincipalDAO kerberosPrincipalDAO;
+
+  /**
+   * KerberosPrincipalHostDAO used to get Kerberos principal details
+   */
+  @Inject
+  private KerberosPrincipalHostDAO kerberosPrincipalHostDAO;
+
   /**
    * Called to execute this action.  Upon invocation, calls
    * {@link org.apache.ambari.server.serveraction.kerberos.KerberosServerAction#processIdentities(java.util.Map)} )}
@@ -129,25 +147,61 @@ public class CreateKeytabFilesServerAction extends KerberosServerAction {
           // If found create th keytab file, else skip it.
           String password = principalPasswordMap.get(evaluatedPrincipal);
 
-          if (password == null) {
-            message = String.format("Failed to create keytab file for %s, missing password", evaluatedPrincipal);
-            actionLog.writeStdErr(message);
-            LOG.error(message);
-            commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
-          } else {
-            // Determine where to store the keytab file.  It should go into a host-specific
-            // directory under the previously determined data directory.
-            File hostDirectory = new File(getDataDirectoryPath(), host);
-
-            // Ensure the host directory exists...
-            if (hostDirectory.exists() || hostDirectory.mkdirs()) {
-              File keytabFile = new File(hostDirectory, DigestUtils.sha1Hex(keytabFilePath));
+          // Determine where to store the keytab file.  It should go into a host-specific
+          // directory under the previously determined data directory.
+          File hostDirectory = new File(getDataDirectoryPath(), host);
+
+          // Ensure the host directory exists...
+          if (hostDirectory.exists() || hostDirectory.mkdirs()) {
+            File keytabFile = new File(hostDirectory, DigestUtils.sha1Hex(keytabFilePath));
+
+            if (password == null) {
+              if (kerberosPrincipalHostDAO.exists(evaluatedPrincipal, host)) {
+                // There is nothing to do for this since it must already exist and we don't want to
+                // regenerate the keytab
+                message = String.format("Skipping keytab file for %s, missing password indicates nothing to do", evaluatedPrincipal);
+                LOG.debug(message);
+              } else {
+                KerberosPrincipalEntity principalEntity = kerberosPrincipalDAO.find(evaluatedPrincipal);
+                String cachedKeytabPath = (principalEntity == null) ? null : principalEntity.getCachedKeytabPath();
+
+                if (cachedKeytabPath == null) {
+                  message = String.format("Failed to create keytab file for %s, missing password", evaluatedPrincipal);
+                  actionLog.writeStdErr(message);
+                  LOG.error(message);
+                  commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
+                } else {
+                  try {
+                    FileUtils.copyFile(new File(cachedKeytabPath), keytabFile);
+                    message = String.format("Using cached keytab file for %s at %s", evaluatedPrincipal, keytabFile.getAbsolutePath());
+                    LOG.debug(message);
+                  } catch (IOException e) {
+                    message = String.format("Failed to use cached keytab file for %s at %s: %s", evaluatedPrincipal, keytabFile.getAbsolutePath(), e.getMessage());
+                    actionLog.writeStdErr(message);
+                    LOG.warn(message);
+                    commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
+                  }
+                }
+              }
+            } else {
               Integer keyNumber = principalKeyNumberMap.get(evaluatedPrincipal);
 
               try {
                 if (operationHandler.createKeytabFile(evaluatedPrincipal, password, keyNumber, keytabFile)) {
                   message = String.format("Successfully created keytab file for %s at %s", evaluatedPrincipal, keytabFile.getAbsolutePath());
                   LOG.debug(message);
+
+                  // If the current identity does not represent a service, store the location of the
+                  // keytab file so it can be reused rather than recreate it.
+                  // Note: for now we are using the keytab's destination directory on the Ambari
+                  // server
+                  KerberosPrincipalEntity principalEntity = kerberosPrincipalDAO.find(evaluatedPrincipal);
+                  if (principalEntity != null) {
+                    if (!principalEntity.isService()) {
+                      principalEntity.setCachedKeytabPath(keytabFilePath);
+                      kerberosPrincipalDAO.merge(principalEntity);
+                    }
+                  }
                 } else {
                   message = String.format("Failed to create keytab file for %s at %s", evaluatedPrincipal, keytabFile.getAbsolutePath());
                   actionLog.writeStdErr(message);
@@ -160,13 +214,13 @@ public class CreateKeytabFilesServerAction extends KerberosServerAction {
                 LOG.error(message, e);
                 commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
               }
-            } else {
-              message = String.format("Failed to create keytab file for %s, the container directory does not exist: %s",
-                  evaluatedPrincipal, hostDirectory.getAbsolutePath());
-              actionLog.writeStdErr(message);
-              LOG.error(message);
-              commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
             }
+          } else {
+            message = String.format("Failed to create keytab file for %s, the container directory does not exist: %s",
+                evaluatedPrincipal, hostDirectory.getAbsolutePath());
+            actionLog.writeStdErr(message);
+            LOG.error(message);
+            commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
           }
         }
       }

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

@@ -18,9 +18,12 @@
 
 package org.apache.ambari.server.serveraction.kerberos;
 
+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.orm.dao.KerberosPrincipalDAO;
+import org.apache.ambari.server.orm.dao.KerberosPrincipalHostDAO;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,6 +41,17 @@ import java.util.concurrent.ConcurrentMap;
 public class CreatePrincipalsServerAction extends KerberosServerAction {
   private final static Logger LOG = LoggerFactory.getLogger(CreatePrincipalsServerAction.class);
 
+  /**
+   * KerberosPrincipalDAO used to set and get Kerberos principal details
+   */
+  @Inject
+  private KerberosPrincipalDAO kerberosPrincipalDAO;
+
+  /**
+   * KerberosPrincipalHostDAO used to get Kerberos principal details
+   */
+  @Inject
+  private KerberosPrincipalHostDAO kerberosPrincipalHostDAO;
 
   /**
    * Called to execute this action.  Upon invocation, calls
@@ -87,62 +101,72 @@ public class CreatePrincipalsServerAction extends KerberosServerAction {
       throws AmbariException {
     CommandReport commandReport = null;
 
-    Map<String, String> principalPasswordMap = getPrincipalPasswordMap(requestSharedDataContext);
-    Map<String, Integer> principalKeyNumberMap = getPrincipalKeyNumberMap(requestSharedDataContext);
+    boolean regenerateKeytabs = "true".equalsIgnoreCase(getCommandParameterValue(getCommandParameters(), REGENERATE_ALL));
 
-    String password = principalPasswordMap.get(evaluatedPrincipal);
+    if (regenerateKeytabs || !kerberosPrincipalHostDAO.exists(evaluatedPrincipal)) {
+      Map<String, String> principalPasswordMap = getPrincipalPasswordMap(requestSharedDataContext);
+      Map<String, Integer> principalKeyNumberMap = getPrincipalKeyNumberMap(requestSharedDataContext);
 
-    if (password == null) {
-      String message = String.format("Creating principal, %s", evaluatedPrincipal);
-      LOG.info(message);
-      actionLog.writeStdOut(message);
+      String password = principalPasswordMap.get(evaluatedPrincipal);
 
-      password = operationHandler.createSecurePassword();
+      if (password == null) {
+        String message = String.format("Creating principal, %s", evaluatedPrincipal);
+        LOG.info(message);
+        actionLog.writeStdOut(message);
 
-      try {
-        if (operationHandler.principalExists(evaluatedPrincipal)) {
-          // Create a new password since we need to know what it is.
-          // A new password/key would have been generated after exporting the keytab anyways.
-          message = String.format("Principal, %s, already exists, setting new password", evaluatedPrincipal);
-          LOG.warn(message);
-          actionLog.writeStdOut(message);
-          Integer keyNumber = operationHandler.setPrincipalPassword(evaluatedPrincipal, password);
-
-          if (keyNumber != null) {
-            principalPasswordMap.put(evaluatedPrincipal, password);
-            principalKeyNumberMap.put(evaluatedPrincipal, keyNumber);
-            message = String.format("Successfully set password for %s", evaluatedPrincipal);
-            LOG.debug(message);
-          } else {
-            message = String.format("Failed to set password for %s - unknown reason", evaluatedPrincipal);
-            LOG.error(message);
-            actionLog.writeStdErr(message);
-            commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
-          }
-        } else {
-          message = String.format("Creating new principal, %s", evaluatedPrincipal);
-          LOG.debug(message);
+        password = operationHandler.createSecurePassword();
 
+        try {
           boolean servicePrincipal = "service".equalsIgnoreCase(identityRecord.get(KerberosActionDataFile.PRINCIPAL_TYPE));
-          Integer keyNumber = operationHandler.createPrincipal(evaluatedPrincipal, password, servicePrincipal);
 
-          if (keyNumber != null) {
-            principalPasswordMap.put(evaluatedPrincipal, password);
-            principalKeyNumberMap.put(evaluatedPrincipal, keyNumber);
-            message = String.format("Successfully created new principal, %s", evaluatedPrincipal);
-            LOG.debug(message);
+          if (operationHandler.principalExists(evaluatedPrincipal)) {
+            // Create a new password since we need to know what it is.
+            // A new password/key would have been generated after exporting the keytab anyways.
+            message = String.format("Principal, %s, already exists, setting new password", evaluatedPrincipal);
+            LOG.warn(message);
+            actionLog.writeStdOut(message);
+            Integer keyNumber = operationHandler.setPrincipalPassword(evaluatedPrincipal, password);
+
+            if (keyNumber != null) {
+              principalPasswordMap.put(evaluatedPrincipal, password);
+              principalKeyNumberMap.put(evaluatedPrincipal, keyNumber);
+              message = String.format("Successfully set password for %s", evaluatedPrincipal);
+              LOG.debug(message);
+            } else {
+              message = String.format("Failed to set password for %s - unknown reason", evaluatedPrincipal);
+              LOG.error(message);
+              actionLog.writeStdErr(message);
+              commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
+            }
           } else {
-            message = String.format("Failed to create principal, %s - unknown reason", evaluatedPrincipal);
-            LOG.error(message);
-            actionLog.writeStdErr(message);
-            commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
+            message = String.format("Creating new principal, %s", evaluatedPrincipal);
+            LOG.debug(message);
+
+            Integer keyNumber = operationHandler.createPrincipal(evaluatedPrincipal, password, servicePrincipal);
+
+            if (keyNumber != null) {
+              principalPasswordMap.put(evaluatedPrincipal, password);
+              principalKeyNumberMap.put(evaluatedPrincipal, keyNumber);
+              message = String.format("Successfully created new principal, %s", evaluatedPrincipal);
+              LOG.debug(message);
+            } else {
+              message = String.format("Failed to create principal, %s - unknown reason", evaluatedPrincipal);
+              LOG.error(message);
+              actionLog.writeStdErr(message);
+              commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
+            }
           }
+
+          if (!kerberosPrincipalDAO.exists(evaluatedPrincipal)) {
+            kerberosPrincipalDAO.create(evaluatedPrincipal, servicePrincipal);
+          }
+
+        } catch (KerberosOperationException e) {
+          message = String.format("Failed to create principal, %s - %s", evaluatedPrincipal, e.getMessage());
+          LOG.error(message, e);
+          actionLog.writeStdErr(message);
+          commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
         }
-      } catch (KerberosOperationException e) {
-        message = String.format("Failed to create principal, %s - %s", evaluatedPrincipal, e.getMessage());
-        LOG.error(message, e);
-        actionLog.writeStdErr(message);
-        commandReport = createCommandReport(1, HostRoleStatus.FAILED, "{}", actionLog.getStdOut(), actionLog.getStdErr());
       }
     }
 

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

@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.serveraction.kerberos;
+
+import com.google.inject.Inject;
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.agent.CommandReport;
+import org.apache.ambari.server.orm.dao.KerberosPrincipalDAO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * DestroyPrincipalsServerAction is a ServerAction implementation that destroys principals as instructed.
+ * <p/>
+ * This class mainly relies on the KerberosServerAction to iterate through metadata identifying
+ * the Kerberos principals that need to be removed from the relevant KDC. For each identity in the
+ * metadata, this implementation's
+ * {@link org.apache.ambari.server.serveraction.kerberos.KerberosServerAction#processIdentity(java.util.Map, String, org.apache.ambari.server.serveraction.kerberos.KerberosOperationHandler, java.util.Map)}
+ * is invoked attempting the removal of the relevant principal.
+ */
+public class DestroyPrincipalsServerAction extends KerberosServerAction {
+  private final static Logger LOG = LoggerFactory.getLogger(DestroyPrincipalsServerAction.class);
+
+  @Inject
+  private KerberosPrincipalDAO kerberosPrincipalDAO;
+
+  /**
+   * Called to execute this action.  Upon invocation, calls
+   * {@link KerberosServerAction#processIdentities(java.util.Map)}
+   * to iterate through the Kerberos identity metadata and call
+   * {@link org.apache.ambari.server.serveraction.kerberos.DestroyPrincipalsServerAction#processIdentities(java.util.Map)}
+   * for each identity to process.
+   *
+   * @param requestSharedDataContext a Map to be used a shared data among all ServerActions related
+   *                                 to a given request
+   * @return a CommandReport indicating the result of this action
+   * @throws org.apache.ambari.server.AmbariException
+   * @throws InterruptedException
+   */
+  @Override
+  public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext) throws
+      AmbariException, InterruptedException {
+    return processIdentities(requestSharedDataContext);
+  }
+
+
+  /**
+   * For each identity, remove the principal from the configured KDC.
+   *
+   * @param identityRecord           a Map containing the data for the current identity record
+   * @param evaluatedPrincipal       a String indicating the relevant principal
+   * @param operationHandler         a KerberosOperationHandler used to perform Kerberos-related
+   *                                 tasks for specific Kerberos implementations
+   *                                 (MIT, Active Directory, etc...)
+   * @param requestSharedDataContext a Map to be used a shared data among all ServerActions related
+   *                                 to a given request
+   * @return a CommandReport, indicating an error condition; or null, indicating a success condition
+   * @throws org.apache.ambari.server.AmbariException if an error occurs while processing the identity record
+   */
+  @Override
+  protected CommandReport processIdentity(Map<String, String> identityRecord, String evaluatedPrincipal,
+                                          KerberosOperationHandler operationHandler,
+                                          Map<String, Object> requestSharedDataContext)
+      throws AmbariException {
+
+    String message = String.format("Destroying identity, %s", evaluatedPrincipal);
+    LOG.info(message);
+    actionLog.writeStdOut(message);
+
+    try {
+      operationHandler.removePrincipal(evaluatedPrincipal);
+    } catch (KerberosOperationException e) {
+      message = String.format("Failed to remove identity for %s from the KDC - %s", evaluatedPrincipal, e.getMessage());
+      LOG.warn(message);
+      actionLog.writeStdErr(message);
+    }
+
+    try {
+      kerberosPrincipalDAO.remove(evaluatedPrincipal);
+    }
+    catch (Throwable t) {
+      message = String.format("Failed to remove identity for %s from the Ambari database - %s", evaluatedPrincipal, t.getMessage());
+      LOG.warn(message);
+      actionLog.writeStdErr(message);
+    }
+
+    // There is no reason to fail this task if an identity was not removed. The cluster will work
+    // just fine if this cleanup process fails.
+    return null;
+  }
+}

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

@@ -89,6 +89,12 @@ public abstract class KerberosServerAction extends AbstractServerAction {
   */
   public static final String KEYTAB_CONTENT_BASE64 = "keytab_content_base64";
 
+  /*
+  * Key used in kerberosCommandParams in ExecutionCommand to indicate whether to generate key keytabs
+  * for all principals ("true") or only those that are missing ("false")
+  */
+  public static final String REGENERATE_ALL = "regenerate_all";
+
   private static final Logger LOG = LoggerFactory.getLogger(KerberosServerAction.class);
 
   /**

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

@@ -66,6 +66,8 @@ public class UpgradeCatalog200 extends AbstractUpgradeCatalog {
   private static final String ALERT_TARGET_STATES_TABLE = "alert_target_states";
   private static final String ALERT_CURRENT_TABLE = "alert_current";
   private static final String ARTIFACT_TABLE = "artifact";
+  private static final String KERBEROS_PRINCIPAL_TABLE = "kerberos_principal";
+  private static final String KERBEROS_PRINCIPAL_HOST_TABLE = "kerberos_principal_host";
 
   /**
    * {@inheritDoc}
@@ -111,6 +113,7 @@ public class UpgradeCatalog200 extends AbstractUpgradeCatalog {
     prepareRollingUpgradesDDL();
     executeAlertDDLUpdates();
     createArtifactTable();
+    createKerberosPrincipalTables();
 
     // add security_type to clusters
     dbAccessor.addColumn("clusters", new DBColumnInfo(
@@ -272,6 +275,23 @@ public class UpgradeCatalog200 extends AbstractUpgradeCatalog {
     dbAccessor.createTable(ARTIFACT_TABLE, columns, "artifact_name", "foreign_keys");
   }
 
+  private void createKerberosPrincipalTables() throws SQLException {
+    ArrayList<DBColumnInfo> columns;
+
+    columns = new ArrayList<DBColumnInfo>();
+    columns.add(new DBColumnInfo("principal_name", String.class, 255, null, false));
+    columns.add(new DBColumnInfo("is_service", Short.class, 1, 1, false));
+    columns.add(new DBColumnInfo("cached_keytab_path", String.class, 255, null, true));
+    dbAccessor.createTable(KERBEROS_PRINCIPAL_TABLE, columns, "principal_name");
+
+    columns = new ArrayList<DBColumnInfo>();
+    columns.add(new DBColumnInfo("principal_name", String.class, 255, null, false));
+    columns.add(new DBColumnInfo("host_name", String.class, 255, null, false));
+    dbAccessor.createTable(KERBEROS_PRINCIPAL_HOST_TABLE, columns, "principal_name", "host_name");
+    dbAccessor.addFKConstraint(KERBEROS_PRINCIPAL_HOST_TABLE, "FK_kerberosprincipalhost_hostname", "host_name", "hosts", "host_name", false);
+    dbAccessor.addFKConstraint(KERBEROS_PRINCIPAL_HOST_TABLE, "FK_kerberosprincipalhost_principalname", "principal_name", KERBEROS_PRINCIPAL_TABLE, "principal_name", false);
+  }
+
   // ----- UpgradeCatalog ----------------------------------------------------
 
   /**

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

@@ -598,6 +598,29 @@ ALTER TABLE users ADD CONSTRAINT FK_users_principal_id FOREIGN KEY (principal_id
 ALTER TABLE groups ADD CONSTRAINT FK_groups_principal_id FOREIGN KEY (principal_id) REFERENCES adminprincipal(principal_id);
 ALTER TABLE clusters ADD CONSTRAINT FK_clusters_resource_id FOREIGN KEY (resource_id) REFERENCES adminresource(resource_id);
 
+-- Kerberos
+CREATE TABLE kerberos_principal (
+  principal_name VARCHAR(255) NOT NULL,
+  is_service SMALLINT NOT NULL DEFAULT 1,
+  cached_keytab_path VARCHAR(255),
+  PRIMARY KEY(principal_name)
+);
+
+CREATE TABLE kerberos_principal_host (
+  principal_name VARCHAR(255) NOT NULL,
+  host_name VARCHAR(255) NOT NULL,
+  PRIMARY KEY(principal_name, host_name)
+);
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_hostname
+FOREIGN KEY (host_name) REFERENCES hosts (host_name) ON DELETE CASCADE;
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_principalname
+FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name) ON DELETE CASCADE;
+-- Kerberos (end)
+
 -- Alerting Framework
 CREATE TABLE alert_definition (
   definition_id BIGINT NOT NULL, 

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

@@ -588,6 +588,29 @@ ALTER TABLE users ADD CONSTRAINT FK_users_principal_id FOREIGN KEY (principal_id
 ALTER TABLE groups ADD CONSTRAINT FK_groups_principal_id FOREIGN KEY (principal_id) REFERENCES adminprincipal(principal_id);
 ALTER TABLE clusters ADD CONSTRAINT FK_clusters_resource_id FOREIGN KEY (resource_id) REFERENCES adminresource(resource_id);
 
+-- Kerberos
+CREATE TABLE kerberos_principal (
+  principal_name VARCHAR2(255) NOT NULL,
+  is_service NUMBER(1) DEFAULT 1 NOT NULL,
+  cached_keytab_path VARCHAR2(255),
+  PRIMARY KEY(principal_name)
+);
+
+CREATE TABLE kerberos_principal_host (
+  principal_name VARCHAR2(255) NOT NULL,
+  host_name VARCHAR2(255) NOT NULL,
+  PRIMARY KEY(principal_name, host_name)
+);
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_hostname
+FOREIGN KEY (host_name) REFERENCES hosts (host_name) ON DELETE CASCADE;
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_principalname
+FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name) ON DELETE CASCADE;
+-- Kerberos (end)
+
 -- Alerting Framework
 CREATE TABLE alert_definition (
   definition_id NUMBER(19) NOT NULL, 

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

@@ -585,6 +585,29 @@ ALTER TABLE serviceconfigmapping ADD CONSTRAINT FK_scvm_config FOREIGN KEY (conf
 ALTER TABLE serviceconfighosts ADD CONSTRAINT  FK_scvhosts_scv FOREIGN KEY (service_config_id) REFERENCES serviceconfig(service_config_id);
 ALTER TABLE clusters ADD CONSTRAINT FK_clusters_resource_id FOREIGN KEY (resource_id) REFERENCES adminresource(resource_id);
 
+-- Kerberos
+CREATE TABLE kerberos_principal (
+  principal_name VARCHAR(255) NOT NULL,
+  is_service SMALLINT NOT NULL DEFAULT 1,
+  cached_keytab_path VARCHAR(255),
+  PRIMARY KEY(principal_name)
+);
+
+CREATE TABLE kerberos_principal_host (
+  principal_name VARCHAR(255) NOT NULL,
+  host_name VARCHAR(255) NOT NULL,
+  PRIMARY KEY(principal_name, host_name)
+);
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_hostname
+FOREIGN KEY (host_name) REFERENCES hosts (host_name) ON DELETE CASCADE;
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_principalname
+FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name) ON DELETE CASCADE;
+-- Kerberos (end)
+
 -- Alerting Framework
 CREATE TABLE alert_definition (
   definition_id BIGINT NOT NULL, 

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

@@ -659,6 +659,31 @@ ALTER TABLE ambari.users ADD CONSTRAINT FK_users_principal_id FOREIGN KEY (princ
 ALTER TABLE ambari.groups ADD CONSTRAINT FK_groups_principal_id FOREIGN KEY (principal_id) REFERENCES ambari.adminprincipal(principal_id);
 ALTER TABLE ambari.clusters ADD CONSTRAINT FK_clusters_resource_id FOREIGN KEY (resource_id) REFERENCES ambari.adminresource(resource_id);
 
+-- Kerberos
+CREATE TABLE ambari.kerberos_principal (
+  principal_name VARCHAR(255) NOT NULL,
+  is_service SMALLINT NOT NULL DEFAULT 1,
+  cached_keytab_path VARCHAR(255),
+  PRIMARY KEY(principal_name)
+);
+GRANT ALL PRIVILEGES ON TABLE ambari.kerberos_principal TO :username;
+
+CREATE TABLE ambari.kerberos_principal_host (
+  principal_name VARCHAR(255) NOT NULL,
+  host_name VARCHAR(255) NOT NULL,
+  PRIMARY KEY(principal_name, host_name)
+);
+GRANT ALL PRIVILEGES ON TABLE ambari.kerberos_principal_host TO :username;
+
+ALTER TABLE ambari.kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_hostname
+FOREIGN KEY (host_name) REFERENCES ambari.hosts (host_name) ON DELETE CASCADE;
+
+ALTER TABLE ambari.kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_principalname
+FOREIGN KEY (principal_name) REFERENCES ambari.kerberos_principal (principal_name) ON DELETE CASCADE;
+-- Kerberos (end)
+
 -- Alerting Framework
 CREATE TABLE ambari.alert_definition (
   definition_id BIGINT NOT NULL, 

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

@@ -163,6 +163,29 @@ ALTER TABLE clusters ADD CONSTRAINT FK_clusters_resource_id FOREIGN KEY (resourc
 ALTER TABLE host_version ADD CONSTRAINT FK_host_version_host_name FOREIGN KEY (host_name) REFERENCES hosts (host_name);
 ALTER TABLE host_version ADD CONSTRAINT FK_host_version_repovers_id FOREIGN KEY (repo_version_id) REFERENCES repo_version (repo_version_id);
 
+-- Kerberos
+CREATE TABLE kerberos_principal (
+  principal_name VARCHAR(255) NOT NULL,
+  is_service SMALLINT NOT NULL DEFAULT 1,
+  cached_keytab_path VARCHAR(255),
+  PRIMARY KEY(principal_name)
+);
+
+CREATE TABLE kerberos_principal_host (
+  principal_name VARCHAR(255) NOT NULL,
+  host_name VARCHAR(255) NOT NULL,
+  PRIMARY KEY(principal_name, host_name)
+);
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_hostname
+FOREIGN KEY (host_name) REFERENCES hosts (host_name) ON DELETE CASCADE;
+
+ALTER TABLE kerberos_principal_host
+ADD CONSTRAINT FK_kerberosprincipalhost_principalname
+FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name) ON DELETE CASCADE;
+-- Kerberos (end)
+
 -- Alerting Framework
 CREATE TABLE alert_definition (
   definition_id BIGINT NOT NULL,

ambari-server/src/main/resources/META-INF/persistence.xml

@@ -43,6 +43,8 @@
     <class>org.apache.ambari.server.orm.entities.HostRoleCommandEntity</class>
     <class>org.apache.ambari.server.orm.entities.HostStateEntity</class>
     <class>org.apache.ambari.server.orm.entities.HostVersionEntity</class>
+    <class>org.apache.ambari.server.orm.entities.KerberosPrincipalEntity</class>
+    <class>org.apache.ambari.server.orm.entities.KerberosPrincipalHostEntity</class>
     <class>org.apache.ambari.server.orm.entities.KeyValueEntity</class>
     <class>org.apache.ambari.server.orm.entities.MemberEntity</class>
     <class>org.apache.ambari.server.orm.entities.MetainfoEntity</class>

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

@@ -42,7 +42,7 @@ class KerberosClient(KerberosScript):
           cached_kinit_executor(status_params.kinit_path_local,
                                 status_params.smoke_user,
                                 status_params.smoke_user_keytab,
-                                status_params.smoke_user,
+                                status_params.smoke_user_principal,
                                 status_params.hostname,
                                 status_params.tmp_dir)
           self.put_structured_out({"securityState": "SECURED_KERBEROS"})
@@ -56,7 +56,7 @@ class KerberosClient(KerberosScript):
       self.put_structured_out({"securityState": "UNSECURED"})
 
   def set_keytab(self, env):
-    KerberosScript.write_keytab_file()
+    self.write_keytab_file()
 
 if __name__ == "__main__":
   KerberosClient().execute()

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

@@ -351,8 +351,7 @@ class KerberosScript(Script):
       return 0, ''
 
 
-  @staticmethod
-  def write_keytab_file():
+  def write_keytab_file(self):
     import params
     import stat
 
@@ -393,3 +392,9 @@ class KerberosScript(Script):
                  mode=mode,
                  owner=owner,
                  group=group)
+
+            principal = get_property_value(item, 'principal')
+            if principal is not None:
+              self.put_structured_out({
+                principal.replace("_HOST", params.hostname): keytab_file_path
+              })

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

@@ -49,6 +49,8 @@ cluster_env = None
 kdc_server_host = None
 cluster_host_info = None
 
+hostname = config['hostname']
+
 kdb5_util_path = 'kdb5_util'
 
 kdamin_pid_path = '/var/run/kadmind.pid'

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

@@ -0,0 +1,32 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+"""
+
+from resource_management import *
+
+config = Script.get_config()
+tmp_dir = Script.get_tmp_dir()
+
+hostname = config['hostname']
+kinit_path_local = functions.get_kinit_path(["/usr/bin", "/usr/kerberos/bin", "/usr/sbin"])
+
+security_enabled = config['configurations']['cluster-env']['security_enabled']
+
+smoke_user_keytab = config['configurations']['cluster-env']['smokeuser_keytab']
+smoke_user = config['configurations']['cluster-env']['smokeuser']
+smoke_user_principal = config['configurations']['cluster-env']['smokeuser_principal_name']

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

@@ -1,144 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.ambari.server.agent;
-
-
-import org.apache.ambari.server.AmbariException;
-import org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFile;
-import org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFileBuilder;
-import org.apache.ambari.server.serveraction.kerberos.KerberosServerAction;
-import org.apache.commons.codec.binary.Base64;
-import org.apache.commons.codec.digest.DigestUtils;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Tests injectKeytab method of HeartBeatHandler
- */
-public class HeartBeatHandlerInjectKeytabTest  {
-
-  String dataDir;
-  
-  @Before
-  public void setup() throws Exception {
-      File temporaryDirectory;
-      File indexFile;
-      KerberosActionDataFileBuilder kerberosActionDataFileBuilder = null;
-
-      try {
-          temporaryDirectory = File.createTempFile(".ambari_", ".d");
-      } catch (IOException e) {
-          throw new AmbariException("Unexpected error", e);
-      }
-
-      // Convert the temporary file into a temporary directory...
-      if (!temporaryDirectory.delete() || !temporaryDirectory.mkdirs()) {
-          throw new AmbariException("Failed to create temporary directory");
-      }
-
-      dataDir = temporaryDirectory.getAbsolutePath();
-
-      indexFile = new File(temporaryDirectory, KerberosActionDataFile.DATA_FILE_NAME);
-      kerberosActionDataFileBuilder = new KerberosActionDataFileBuilder(indexFile);
-
-      kerberosActionDataFileBuilder.addRecord("c6403.ambari.apache.org", "HDFS", "DATANODE",
-              "dn/_HOST@_REALM", "service", "hdfs-site/dfs.namenode.kerberos.principal",
-              "/etc/security/keytabs/dn.service.keytab",
-              "hdfs", "r", "hadoop", "", "hdfs-site/dfs.namenode.keytab.file");
-
-      kerberosActionDataFileBuilder.close();
-      File hostDirectory = new File(dataDir, "c6403.ambari.apache.org");
-
-      // Ensure the host directory exists...
-      if (hostDirectory.exists() || hostDirectory.mkdirs()) {
-          File file = new File(hostDirectory, DigestUtils.sha1Hex("/etc/security/keytabs/dn.service.keytab"));
-          if (!file.exists()) {
-              file.createNewFile();
-          }
-
-          FileWriter fw = new FileWriter(file.getAbsoluteFile());
-          BufferedWriter bw = new BufferedWriter(fw);
-          bw.write("hello");
-          bw.close();
-      }
-  }
-
-    @Test
-    public void testInjectKeytabApplicableHost() throws Exception {
-
-        ExecutionCommand executionCommand = new ExecutionCommand();
-
-        Map<String, String> hlp = new HashMap<String, String>();
-        hlp.put("custom_command", "SET_KEYTAB");
-        executionCommand.setHostLevelParams(hlp);
-
-        Map<String, String> commandparams = new HashMap<String, String>();
-        commandparams.put(KerberosServerAction.DATA_DIRECTORY, dataDir);
-        executionCommand.setCommandParams(commandparams);
-
-        HeartBeatHandler.injectKeytab(executionCommand, "c6403.ambari.apache.org");
-
-        List<Map<String, String>> kcp = executionCommand.getKerberosCommandParams();
-
-        Assert.assertEquals("c6403.ambari.apache.org", kcp.get(0).get(KerberosActionDataFile.HOSTNAME));
-        Assert.assertEquals("HDFS", kcp.get(0).get(KerberosActionDataFile.SERVICE));
-        Assert.assertEquals("DATANODE", kcp.get(0).get(KerberosActionDataFile.COMPONENT));
-        Assert.assertEquals("dn/_HOST@_REALM", kcp.get(0).get(KerberosActionDataFile.PRINCIPAL));
-        Assert.assertEquals("hdfs-site/dfs.namenode.kerberos.principal", kcp.get(0).get(KerberosActionDataFile.PRINCIPAL_CONFIGURATION));
-        Assert.assertEquals("/etc/security/keytabs/dn.service.keytab", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_PATH));
-        Assert.assertEquals("hdfs", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME));
-        Assert.assertEquals("r", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS));
-        Assert.assertEquals("hadoop", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME));
-        Assert.assertEquals("", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS));
-        Assert.assertEquals("hdfs-site/dfs.namenode.keytab.file", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION));
-
-        Assert.assertEquals(Base64.encodeBase64String("hello".getBytes()), kcp.get(0).get(KerberosServerAction.KEYTAB_CONTENT_BASE64));
-
-    }
-
-  @Test
-  public void testInjectKeytabNotApplicableHost() throws Exception {
-      ExecutionCommand executionCommand = new ExecutionCommand();
-
-      Map<String, String> hlp = new HashMap<String, String>();
-      hlp.put("custom_command", "SET_KEYTAB");
-      executionCommand.setHostLevelParams(hlp);
-
-      Map<String, String> commandparams = new HashMap<String, String>();
-      commandparams.put(KerberosServerAction.DATA_DIRECTORY, dataDir);
-      executionCommand.setCommandParams(commandparams);
-
-      HeartBeatHandler.injectKeytab(executionCommand, "c6400.ambari.apache.org");
-
-      List<Map<String, String>> kcp = executionCommand.getKerberosCommandParams();
-      Assert.assertTrue(kcp.isEmpty());
-
-  }
-
-}

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

@@ -46,6 +46,9 @@ import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -81,6 +84,9 @@ import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.orm.OrmTestHelper;
 import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
 import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
+import org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFile;
+import org.apache.ambari.server.serveraction.kerberos.KerberosActionDataFileBuilder;
+import org.apache.ambari.server.serveraction.kerberos.KerberosServerAction;
 import org.apache.ambari.server.state.Alert;
 import org.apache.ambari.server.state.AlertState;
 import org.apache.ambari.server.state.Cluster;
@@ -100,10 +106,14 @@ import org.apache.ambari.server.state.svccomphost.ServiceComponentHostInstallEve
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStartEvent;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostUpgradeEvent;
 import org.apache.ambari.server.utils.StageUtils;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.codec.digest.DigestUtils;
 import org.codehaus.jackson.JsonGenerationException;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -133,6 +143,9 @@ public class TestHeartbeatHandler {
 
   private UnitOfWork unitOfWork;
 
+  @Rule
+  public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
 
   @Before
   public void setup() throws Exception {
@@ -2468,6 +2481,93 @@ public class TestHeartbeatHandler {
 
     entity = dao.findByStackAndVersion("HDP-0.1", "2.2.1.0-2222");
     Assert.assertNotNull(entity);
+  }
+
+  @Test
+  public void testInjectKeytabApplicableHost() throws Exception {
+    List<Map<String, String>> kcp = testInjectKeytab("c6403.ambari.apache.org");
+
+    Assert.assertNotNull(kcp);
+    Assert.assertEquals(1, kcp.size());
+    Assert.assertEquals("c6403.ambari.apache.org", kcp.get(0).get(KerberosActionDataFile.HOSTNAME));
+    Assert.assertEquals("HDFS", kcp.get(0).get(KerberosActionDataFile.SERVICE));
+    Assert.assertEquals("DATANODE", kcp.get(0).get(KerberosActionDataFile.COMPONENT));
+    Assert.assertEquals("dn/_HOST@_REALM", kcp.get(0).get(KerberosActionDataFile.PRINCIPAL));
+    Assert.assertEquals("hdfs-site/dfs.namenode.kerberos.principal", kcp.get(0).get(KerberosActionDataFile.PRINCIPAL_CONFIGURATION));
+    Assert.assertEquals("/etc/security/keytabs/dn.service.keytab", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_PATH));
+    Assert.assertEquals("hdfs", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME));
+    Assert.assertEquals("r", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS));
+    Assert.assertEquals("hadoop", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME));
+    Assert.assertEquals("", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS));
+    Assert.assertEquals("hdfs-site/dfs.namenode.keytab.file", kcp.get(0).get(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION));
+
+    Assert.assertEquals(Base64.encodeBase64String("hello".getBytes()), kcp.get(0).get(KerberosServerAction.KEYTAB_CONTENT_BASE64));
 
   }
+
+  @Test
+  public void testInjectKeytabNotApplicableHost() throws Exception {
+    List<Map<String, String>> kcp = testInjectKeytab("c6401.ambari.apache.org");
+    Assert.assertNotNull(kcp);
+    Assert.assertTrue(kcp.isEmpty());
+  }
+
+  private List<Map<String, String>> testInjectKeytab(String targetHost) throws Exception {
+
+    ExecutionCommand executionCommand = new ExecutionCommand();
+
+    Map<String, String> hlp = new HashMap<String, String>();
+    hlp.put("custom_command", "SET_KEYTAB");
+    executionCommand.setHostLevelParams(hlp);
+
+    Map<String, String> commandparams = new HashMap<String, String>();
+    commandparams.put(KerberosServerAction.DATA_DIRECTORY, createTestKeytabData().getAbsolutePath());
+    executionCommand.setCommandParams(commandparams);
+
+    ActionQueue aq = new ActionQueue();
+
+    final HostRoleCommand command = new HostRoleCommand(DummyHostname1,
+        Role.DATANODE, null, null);
+
+    ActionManager am = getMockActionManager();
+    expect(am.getTasks(anyObject(List.class))).andReturn(
+        new ArrayList<HostRoleCommand>() {{
+          add(command);
+        }});
+    replay(am);
+
+    getHeartBeatHandler(am, aq).injectKeytab(executionCommand, targetHost);
+
+    return executionCommand.getKerberosCommandParams();
+  }
+
+
+  private File createTestKeytabData() throws Exception {
+    File dataDirectory = temporaryFolder.newFolder();
+    File indexFile = new File(dataDirectory, KerberosActionDataFile.DATA_FILE_NAME);
+    KerberosActionDataFileBuilder kerberosActionDataFileBuilder = new KerberosActionDataFileBuilder(indexFile);
+    File hostDirectory = new File(dataDirectory, "c6403.ambari.apache.org");
+
+    File keytabFile;
+    if(hostDirectory.mkdirs())
+      keytabFile = new File(hostDirectory, DigestUtils.sha1Hex("/etc/security/keytabs/dn.service.keytab"));
+    else
+      throw new Exception("Failed to create " + hostDirectory.getAbsolutePath());
+
+    kerberosActionDataFileBuilder.addRecord("c6403.ambari.apache.org", "HDFS", "DATANODE",
+        "dn/_HOST@_REALM", "service", "hdfs-site/dfs.namenode.kerberos.principal",
+        "/etc/security/keytabs/dn.service.keytab",
+        "hdfs", "r", "hadoop", "", "hdfs-site/dfs.namenode.keytab.file");
+
+    kerberosActionDataFileBuilder.close();
+
+    // Ensure the host directory exists...
+    FileWriter fw = new FileWriter(keytabFile);
+    BufferedWriter bw = new BufferedWriter(fw);
+    bw.write("hello");
+    bw.close();
+
+    return dataDirectory;
+  }
+
 }

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

@@ -50,7 +50,6 @@ import org.apache.ambari.server.state.Config;
 import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
-import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.SecurityState;
 import org.apache.ambari.server.state.SecurityType;
 import org.apache.ambari.server.state.Service;
@@ -160,7 +159,6 @@ public class KerberosHelperTest extends EasyMockSupport {
         bind(SecurityHelper.class).toInstance(createNiceMock(SecurityHelper.class));
         bind(OsFamily.class).toInstance(createNiceMock(OsFamily.class));
         bind(AmbariCustomCommandExecutionHelper.class).toInstance(createNiceMock(AmbariCustomCommandExecutionHelper.class));
-        bind(MaintenanceStateHelper.class).toInstance(createNiceMock(MaintenanceStateHelper.class));
         bind(AmbariManagementController.class).toInstance(createNiceMock(AmbariManagementController.class));
         bind(AmbariMetaInfo.class).toInstance(metaInfo);
         bind(ActionManager.class).toInstance(createNiceMock(ActionManager.class));
@@ -401,10 +399,6 @@ public class KerberosHelperTest extends EasyMockSupport {
     // TODO: (rlevas) Remove when AMBARI 9121 is complete
     expect(krb5ConfConfig.getProperties()).andReturn(krb5ConfProperties).once();
 
-    final MaintenanceStateHelper maintenanceStateHelper = injector.getInstance(MaintenanceStateHelper.class);
-    expect(maintenanceStateHelper.getEffectiveState(anyObject(ServiceComponentHost.class)))
-        .andReturn(MaintenanceState.OFF).anyTimes();
-
     final Cluster cluster = createNiceMock(Cluster.class);
     expect(cluster.getSecurityType()).andReturn(SecurityType.KERBEROS).once();
     expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).once();
@@ -445,6 +439,9 @@ public class KerberosHelperTest extends EasyMockSupport {
           }
         })
         .once();
+    expect(clusters.getHost("host1"))
+        .andReturn(host)
+        .once();
 
     final AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class);
     expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))
@@ -621,7 +618,6 @@ public class KerberosHelperTest extends EasyMockSupport {
 
     final Host host = createNiceMock(Host.class);
     expect(host.getHostName()).andReturn("host1").once();
-    expect(host.getState()).andReturn(HostState.HEALTHY).once();
 
     final Service service1 = createStrictMock(Service.class);
     expect(service1.getName()).andReturn("SERVICE1").anyTimes();
@@ -654,10 +650,6 @@ public class KerberosHelperTest extends EasyMockSupport {
     // TODO: (rlevas) Remove when AMBARI 9121 is complete
     expect(krb5ConfConfig.getProperties()).andReturn(krb5ConfProperties).once();
 
-    final MaintenanceStateHelper maintenanceStateHelper = injector.getInstance(MaintenanceStateHelper.class);
-    expect(maintenanceStateHelper.getEffectiveState(anyObject(ServiceComponentHost.class)))
-        .andReturn(MaintenanceState.OFF).anyTimes();
-
     final Cluster cluster = createNiceMock(Cluster.class);
     expect(cluster.getSecurityType()).andReturn(SecurityType.NONE).once();
     expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).once();
@@ -809,6 +801,11 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(requestStageContainer.getId()).andReturn(1L).once();
     requestStageContainer.addStages(anyObject(List.class));
     expectLastCall().once();
+    // Destroy Principals Stage
+    expect(requestStageContainer.getLastStageId()).andReturn(2L).anyTimes();
+    expect(requestStageContainer.getId()).andReturn(1L).once();
+    requestStageContainer.addStages(anyObject(List.class));
+    expectLastCall().once();
     // TODO: Add more of these when more stages are added.
     // Clean-up/Finalize Stage
     expect(requestStageContainer.getLastStageId()).andReturn(3L).anyTimes();
@@ -879,10 +876,6 @@ public class KerberosHelperTest extends EasyMockSupport {
     // TODO: (rlevas) Remove when AMBARI 9121 is complete
     expect(krb5ConfConfig.getProperties()).andReturn(krb5ConfProperties).once();
 
-    final MaintenanceStateHelper maintenanceStateHelper = injector.getInstance(MaintenanceStateHelper.class);
-    expect(maintenanceStateHelper.getEffectiveState(anyObject(ServiceComponentHost.class)))
-        .andReturn(MaintenanceState.OFF).anyTimes();
-
     final Cluster cluster = createNiceMock(Cluster.class);
     expect(cluster.getSecurityType()).andReturn(SecurityType.KERBEROS).once();
     expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).once();
@@ -923,6 +916,9 @@ public class KerberosHelperTest extends EasyMockSupport {
           }
         })
         .once();
+    expect(clusters.getHost("host1"))
+        .andReturn(host)
+        .once();
 
     final AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class);
     expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))
@@ -1161,10 +1157,6 @@ public class KerberosHelperTest extends EasyMockSupport {
     // TODO: (rlevas) Remove when AMBARI 9121 is complete
     expect(krb5ConfConfig.getProperties()).andReturn(krb5ConfProperties).once();
 
-    final MaintenanceStateHelper maintenanceStateHelper = injector.getInstance(MaintenanceStateHelper.class);
-    expect(maintenanceStateHelper.getEffectiveState(anyObject(ServiceComponentHost.class)))
-        .andReturn(MaintenanceState.OFF).anyTimes();
-
     final Cluster cluster = createNiceMock(Cluster.class);
     expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).once();
     expect(cluster.getDesiredConfigByType("kerberos-env")).andReturn(kerberosEnvConfig).once();
@@ -1205,6 +1197,9 @@ public class KerberosHelperTest extends EasyMockSupport {
           }
         })
         .once();
+    expect(clusters.getHost("host1"))
+        .andReturn(host)
+        .once();
 
     final AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class);
     expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))

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

@@ -135,6 +135,8 @@ public class UpgradeCatalog200Test {
     Capture<DBAccessor.DBColumnInfo> dataValueColumnCapture = new Capture<DBAccessor.DBColumnInfo>();
     Capture<List<DBAccessor.DBColumnInfo>> alertTargetStatesCapture = new Capture<List<DBAccessor.DBColumnInfo>>();
     Capture<List<DBAccessor.DBColumnInfo>> artifactCapture = new Capture<List<DBAccessor.DBColumnInfo>>();
+    Capture<List<DBAccessor.DBColumnInfo>> kerberosPrincipalCapture = new Capture<List<DBAccessor.DBColumnInfo>>();
+    Capture<List<DBAccessor.DBColumnInfo>> kerberosPrincipalHostCapture = new Capture<List<DBAccessor.DBColumnInfo>>();
 
     Capture<List<DBAccessor.DBColumnInfo>> upgradeCapture = new Capture<List<DBAccessor.DBColumnInfo>>();
     Capture<List<DBAccessor.DBColumnInfo>> upgradeGroupCapture = new Capture<List<DBAccessor.DBColumnInfo>>();
@@ -211,6 +213,18 @@ public class UpgradeCatalog200Test {
     dbAccessor.createTable(eq("artifact"), capture(artifactCapture),
         eq("artifact_name"), eq("foreign_keys"));
 
+    // kerberos_principal
+    dbAccessor.createTable(eq("kerberos_principal"), capture(kerberosPrincipalCapture),
+        eq("principal_name"));
+
+    // kerberos_principal_host
+    dbAccessor.createTable(eq("kerberos_principal_host"), capture(kerberosPrincipalHostCapture),
+        eq("principal_name"), eq("host_name"));
+    dbAccessor.addFKConstraint(eq("kerberos_principal_host"), eq("FK_kerberosprincipalhost_hostname"),
+        eq("host_name"), eq("hosts"), eq("host_name"), eq(false));
+    dbAccessor.addFKConstraint(eq("kerberos_principal_host"), eq("FK_kerberosprincipalhost_principalname"),
+        eq("principal_name"), eq("kerberos_principal"), eq("principal_name"), eq(false));
+
     setViewInstancePropertyExpectations(dbAccessor, valueColumnCapture);
     setViewInstanceDataExpectations(dbAccessor, dataValueColumnCapture);
 
@@ -282,6 +296,11 @@ public class UpgradeCatalog200Test {
     List<DBAccessor.DBColumnInfo> artifactColumns = artifactCapture.getValue();
     testCreateArtifactTable(artifactColumns);
 
+    // verify kerberos_principal columns
+    testCreateKerberosPrincipalTable(kerberosPrincipalCapture.getValue());
+
+    // verify kerberos_principal_host columns
+    testCreateKerberosPrincipalHostTable(kerberosPrincipalHostCapture.getValue());
 
     // Verify capture group sizes
     assertEquals(7, clusterVersionCapture.getValue().size());
@@ -606,4 +625,47 @@ public class UpgradeCatalog200Test {
       }
     }
   }
+
+  private void testCreateKerberosPrincipalTable(List<DBColumnInfo> columns) {
+    assertEquals(3, columns.size());
+    for (DBColumnInfo column : columns) {
+      if (column.getName().equals("principal_name")) {
+        assertNull(column.getDefaultValue());
+        assertEquals(String.class, column.getType());
+        assertEquals(255, (int) column.getLength());
+        assertEquals(false, column.isNullable());
+      } else if (column.getName().equals("is_service")) {
+        assertEquals(1, column.getDefaultValue());
+        assertEquals(Short.class, column.getType());
+        assertEquals(1, (int) column.getLength());
+        assertEquals(false, column.isNullable());
+      } else if (column.getName().equals("cached_keytab_path")) {
+        assertNull(column.getDefaultValue());
+        assertEquals(String.class, column.getType());
+        assertEquals(255, (int) column.getLength());
+        assertEquals(true, column.isNullable());
+      } else {
+        fail("unexpected column name");
+      }
+    }
+  }
+
+  private void testCreateKerberosPrincipalHostTable(List<DBColumnInfo> columns) {
+    assertEquals(2, columns.size());
+    for (DBColumnInfo column : columns) {
+      if (column.getName().equals("principal_name")) {
+        assertNull(column.getDefaultValue());
+        assertEquals(String.class, column.getType());
+        assertEquals(255, (int) column.getLength());
+        assertEquals(false, column.isNullable());
+      } else if (column.getName().equals("host_name")) {
+        assertNull(column.getDefaultValue());
+        assertEquals(String.class, column.getType());
+        assertEquals(255, (int) column.getLength());
+        assertEquals(false, column.isNullable());
+      } else {
+        fail("unexpected column name");
+      }
+    }
+  }
 }