AMBARI-9749. Kerberos: check kerb task should delete smoke user principal (rlevas)

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

@@ -34,6 +34,7 @@ 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.Role;
 import org.apache.ambari.server.RoleCommand;
 import org.apache.ambari.server.ServiceComponentHostNotFoundException;
 import org.apache.ambari.server.ServiceComponentNotFoundException;
@@ -451,29 +452,37 @@ public class HeartBeatHandler {
         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()) &&
+      // If the report indicates the keytab file was successfully transferred to a host or removed
+      // from a host, record this for future reference
+      if (Service.Type.KERBEROS.name().equalsIgnoreCase(report.getServiceName()) &&
+          Role.KERBEROS_CLIENT.name().equalsIgnoreCase(report.getRole()) &&
           RoleCommand.CUSTOM_COMMAND.name().equalsIgnoreCase(report.getRoleCommand()) &&
-          "SET_KEYTAB".equalsIgnoreCase(report.getCustomCommand()) &&
           RequestExecution.Status.COMPLETED.name().equalsIgnoreCase(report.getStatus())) {
 
-        WriteKeytabsStructuredOut writeKeytabsStructuredOut;
-        try {
-          writeKeytabsStructuredOut = gson.fromJson(report.getStructuredOut(), WriteKeytabsStructuredOut.class);
-        } catch (JsonSyntaxException ex) {
-          //Json structure was incorrect do nothing, pass this data further for processing
-          writeKeytabsStructuredOut = null;
-        }
+        String customCommand = report.getCustomCommand();
+
+        boolean adding = "SET_KEYTAB".equalsIgnoreCase(customCommand);
+        if (adding || "REMOVE_KEYTAB".equalsIgnoreCase(customCommand)) {
+          WriteKeytabsStructuredOut writeKeytabsStructuredOut;
+          try {
+            writeKeytabsStructuredOut = gson.fromJson(report.getStructuredOut(), WriteKeytabsStructuredOut.class);
+          } catch (JsonSyntaxException ex) {
+            //Json structure was incorrect do nothing, pass this data further for processing
+            writeKeytabsStructuredOut = null;
+          }
 
-        if (writeKeytabsStructuredOut != null) {
-          Map<String, String> keytabs = writeKeytabsStructuredOut.getKeytabs();
-          if (keytabs != null) {
-            for (Map.Entry<String, String> entry : keytabs.entrySet()) {
-              String principal = entry.getKey();
-              if (!kerberosPrincipalHostDAO.exists(principal, hostname)) {
-                kerberosPrincipalHostDAO.create(principal, hostname);
+          if (writeKeytabsStructuredOut != null) {
+            Map<String, String> keytabs = writeKeytabsStructuredOut.getKeytabs();
+            if (keytabs != null) {
+              for (Map.Entry<String, String> entry : keytabs.entrySet()) {
+                String principal = entry.getKey();
+                if (!kerberosPrincipalHostDAO.exists(principal, hostname)) {
+                  if (adding) {
+                    kerberosPrincipalHostDAO.create(principal, hostname);
+                  } else if ("_REMOVED_".equalsIgnoreCase(entry.getValue())) {
+                    kerberosPrincipalHostDAO.remove(principal, hostname);
+                  }
+                }
               }
             }
           }
@@ -764,12 +773,15 @@ public class HeartBeatHandler {
           case EXECUTION_COMMAND: {
             ExecutionCommand ec = (ExecutionCommand)ac;
             Map<String, String> hlp = ec.getHostLevelParams();
-            if ((hlp != null) && "SET_KEYTAB".equals(hlp.get("custom_command")))   {
-              LOG.info("SET_KEYTAB called") ;
-              try {
-                injectKeytab(ec, hostname);
-              } catch (IOException e) {
-                throw new AmbariException("Could not inject keytab into command", e);
+            if (hlp != null) {
+              String customCommand = hlp.get("custom_command");
+              if ("SET_KEYTAB".equalsIgnoreCase(customCommand) || "REMOVE_KEYTAB".equalsIgnoreCase(customCommand)) {
+                LOG.info(String.format("%s called", customCommand));
+                try {
+                  injectKeytab(ec, customCommand, hostname);
+                } catch (IOException e) {
+                  throw new AmbariException("Could not inject keytab into command", e);
+                }
               }
             }
             response.addExecutionCommand((ExecutionCommand) ac);
@@ -1020,14 +1032,11 @@ public class HeartBeatHandler {
    * any keytab details and associated data exists for the target host.
    *
    * @param ec the ExecutionCommand to update
+   * @param command a name of the relevant keytab command
    * @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;
-    }
+  void injectKeytab(ExecutionCommand ec, String command, String targetHost) throws AmbariException {
     List<Map<String, String>> kcp = ec.getKerberosCommandParams();
     String dataDir = ec.getCommandParams().get(KerberosServerAction.DATA_DIRECTORY);
     KerberosActionDataFileReader reader = null;
@@ -1035,40 +1044,54 @@ public class HeartBeatHandler {
     try {
       reader = new KerberosActionDataFileReader(new File(dataDir, KerberosActionDataFile.DATA_FILE_NAME));
 
-      for(Map<String, String> record : reader) {
+      for (Map<String, String> record : reader) {
         String hostName = record.get(KerberosActionDataFile.HOSTNAME);
 
         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);
+
+          if ("SET_KEYTAB".equalsIgnoreCase(command)) {
+            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);
+              }
             }
+          } else if ("REMOVE_KEYTAB".equalsIgnoreCase(command)) {
+            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.KEYTAB_FILE_PATH, record.get(KerberosActionDataFile.KEYTAB_FILE_PATH));
+
+            kcp.add(keytabMap);
           }
         }
       }

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

@@ -97,6 +97,7 @@ import org.apache.ambari.server.security.ldap.AmbariLdapDataPopulator;
 import org.apache.ambari.server.security.ldap.LdapBatchDto;
 import org.apache.ambari.server.security.ldap.LdapSyncDto;
 import org.apache.ambari.server.serveraction.kerberos.KerberosInvalidConfigurationException;
+import org.apache.ambari.server.serveraction.kerberos.KerberosOperationException;
 import org.apache.ambari.server.stageplanner.RoleGraph;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
@@ -1310,7 +1311,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
 
         if (!isStateTransitionValid) {
           LOG.warn(
-              "Invalid cluster provisioning state {} cannot be set on the cluster {} because the current state is {}",
+              "Invalid cluster provisioning 2state {} cannot be set on the cluster {} because the current state is {}",
               provisioningState, request.getClusterName(), oldProvisioningState);
 
           throw new AmbariException("Invalid transition for"
@@ -1365,7 +1366,11 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
       // if any custom operations are valid and requested, the process of executing them should be initiated,
       // most of the validation logic will be left to the KerberosHelper to avoid polluting the controller
       if (kerberosHelper.shouldExecuteCustomOperations(securityType, requestProperties)) {
-        requestStageContainer = kerberosHelper.executeCustomOperations(cluster, requestProperties, requestStageContainer);
+        try {
+          requestStageContainer = kerberosHelper.executeCustomOperations(cluster, requestProperties, requestStageContainer);
+        } catch (KerberosOperationException e) {
+          throw new IllegalArgumentException(e.getMessage(), e);
+        }
       } else if (cluster.getSecurityType() != securityType) {
         LOG.info("Received cluster security type change request from {} to {}",
             cluster.getSecurityType().name(), securityType.name());
@@ -1374,7 +1379,11 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
           // Since the security state of the cluster has changed, invoke toggleKerberos to handle
           // adding or removing Kerberos from the cluster. This may generate multiple stages
           // or not depending the current state of the cluster.
-          requestStageContainer = kerberosHelper.toggleKerberos(cluster, securityType, requestStageContainer);
+          try {
+            requestStageContainer = kerberosHelper.toggleKerberos(cluster, securityType, requestStageContainer);
+          } catch (KerberosOperationException e) {
+            throw new IllegalArgumentException(e.getMessage(), e);
+          }
         } else {
           throw new IllegalArgumentException(String.format("Unexpected security type encountered: %s", securityType.name()));
         }
@@ -2143,7 +2152,11 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
           componentFilter.add(scHost.getServiceComponentName());
         }
 
-        kerberosHelper.ensureIdentities(cluster, serviceFilter, null, requestStages);
+        try {
+          kerberosHelper.ensureIdentities(cluster, serviceFilter, null, requestStages);
+        } catch (KerberosOperationException e) {
+          throw new IllegalArgumentException(e.getMessage(), e);
+        }
       }
 
       List<Stage> stages = requestStages.getStages();
@@ -2892,12 +2905,13 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     // If the request is to perform the Kerberos service check, set up the stages to
     // ensure that the (cluster-level) smoke user principal and keytab is available on all hosts
     if (Role.KERBEROS_SERVICE_CHECK.name().equals(actionRequest.getCommandName())) {
-      Map<String, Collection<String>> serviceComponentFilter = new HashMap<String, Collection<String>>();
-      Collection<String> identityFilter = Arrays.asList("/smokeuser");
-
-      serviceComponentFilter.put("KERBEROS", null);
-
-      requestStageContainer = kerberosHelper.ensureIdentities(cluster, serviceComponentFilter, identityFilter, requestStageContainer);
+      try {
+        requestStageContainer = kerberosHelper.ensureIdentities(cluster,
+            Collections.<String, Collection<String>>singletonMap(Service.Type.KERBEROS.name(), null),
+            Collections.singleton("/smokeuser"), requestStageContainer);
+      } catch (KerberosOperationException e) {
+        throw new IllegalArgumentException(e.getMessage(), e);
+      }
     }
 
     ExecuteCommandJson jsons = customCommandExecutionHelper.getCommandJson(
@@ -2927,6 +2941,20 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
       requestStageContainer.addStages(stages);
     }
 
+    // If the request is to perform the Kerberos service check and (Kerberos) security is not enabled,
+    // delete that the (cluster-level) smoke user principal and keytab that was created for the
+    // service check
+    if (Role.KERBEROS_SERVICE_CHECK.name().equals(actionRequest.getCommandName()) &&
+        !kerberosHelper.isClusterKerberosEnabled(cluster)) {
+      try {
+        requestStageContainer = kerberosHelper.deleteIdentities(cluster,
+            Collections.<String, Collection<String>>singletonMap(Service.Type.KERBEROS.name(), null),
+            Collections.singleton("/smokeuser"), requestStageContainer);
+      } catch (KerberosOperationException e) {
+        throw new IllegalArgumentException(e.getMessage(), e);
+      }
+    }
+
     requestStageContainer.persist();
     return requestStageContainer.getRequestStatusResponse();
   }

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

@@ -108,7 +108,6 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.TreeMap;
 
 public class KerberosHelper {
   private static final String BASE_LOG_DIR = "/tmp/ambari";
@@ -170,11 +169,6 @@ public class KerberosHelper {
    * the security type is NONE, an attempt will be made to disable Kerberos; otherwise, no operations
    * will be performed.
    * <p/>
-   * It is expected that the "krb5-conf" configuration type is available.  It is used to obtain
-   * information about the relevant KDC.  For example, the "kdc_type" property is used to determine
-   * what type of KDC is being used so that the appropriate actions maybe taken in order interact
-   * with it properly.
-   * <p/>
    * It is expected tht the "kerberos-env" configuration type is available.   It is used to obtain
    * information about the Kerberos configuration, generally specific to the KDC being used.
    * <p/>
@@ -189,17 +183,15 @@ public class KerberosHelper {
    * @return the updated or a new RequestStageContainer containing the stages that need to be
    * executed to complete this task; or null if no stages need to be executed.
    * @throws AmbariException
+   * @throws KerberosInvalidConfigurationException if an issue occurs trying to get the
+   * Kerberos-specific configuration details
+   * @throws KerberosOperationException
    */
   public RequestStageContainer toggleKerberos(Cluster cluster, SecurityType securityType,
                                               RequestStageContainer requestStageContainer)
-      throws AmbariException {
+      throws AmbariException, KerberosOperationException {
 
-    KerberosDetails kerberosDetails;
-    try {
-      kerberosDetails = getKerberosDetails(cluster);
-    } catch (KerberosInvalidConfigurationException e) {
-      throw new IllegalArgumentException(e.getMessage(), e);
-    }
+    KerberosDetails kerberosDetails = getKerberosDetails(cluster);
 
     // Update KerberosDetails with the new security type - the current one in the cluster is the "old" value
     kerberosDetails.setSecurityType(securityType);
@@ -227,10 +219,13 @@ public class KerberosHelper {
    *                              if null a new RequestStageContainer will be created.   @return the updated or a new RequestStageContainer containing the stages that need to be
    *                              executed to complete this task; or null if no stages need to be executed.
    * @throws AmbariException
+   * @throws KerberosOperationException
+   * @throws KerberosInvalidConfigurationException if an issue occurs trying to get the
+   * Kerberos-specific configuration details
    */
   public RequestStageContainer executeCustomOperations(Cluster cluster, Map<String, String> requestProperties,
                                                        RequestStageContainer requestStageContainer)
-      throws AmbariException {
+      throws AmbariException, KerberosOperationException {
 
     if (requestProperties != null) {
 
@@ -245,18 +240,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()));
               }
 
-              try {
-                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));
-                }
-              } catch (KerberosInvalidConfigurationException e) {
-                throw new IllegalArgumentException(e.getMessage(), e);
+              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;
@@ -278,7 +269,7 @@ public class KerberosHelper {
    * No configurations will be altered as a result of this operation, however principals and keytabs
    * may be updated or created.
    * <p/>
-   * It is expected that the "krb5-conf" configuration type is available.  It is used to obtain
+   * It is expected that the "kerberos-env" configuration type is available.  It is used to obtain
    * information about the relevant KDC.  For example, the "kdc_type" property is used to determine
    * what type of KDC is being used so that the appropriate actions maybe taken in order interact
    * with it properly.
@@ -298,16 +289,52 @@ public class KerberosHelper {
    * @return the updated or a new RequestStageContainer containing the stages that need to be
    * executed to complete this task; or null if no stages need to be executed.
    * @throws AmbariException
+   * @throws KerberosOperationException
+   * @throws KerberosInvalidConfigurationException if an issue occurs trying to get the
+   * Kerberos-specific configuration details
    */
   public RequestStageContainer ensureIdentities(Cluster cluster, Map<String, ? extends Collection<String>> serviceComponentFilter,
                                                 Collection<String> identityFilter, RequestStageContainer requestStageContainer)
-      throws AmbariException {
-    try {
-      return handle(cluster, getKerberosDetails(cluster), serviceComponentFilter, identityFilter,
-          requestStageContainer, new CreatePrincipalsAndKeytabsHandler(false));
-    } catch (KerberosInvalidConfigurationException e) {
-      throw new IllegalArgumentException(e.getMessage(), e);
-    }
+      throws AmbariException, KerberosOperationException {
+    return handle(cluster, getKerberosDetails(cluster), serviceComponentFilter, identityFilter,
+        requestStageContainer, new CreatePrincipalsAndKeytabsHandler(false));
+ }
+
+  /**
+   * Deletes the set of filtered principals and keytabs from the cluster.
+   * <p/>
+   * No configurations will be altered as a result of this operation, however principals and keytabs
+   * may be removed.
+   * <p/>
+   * It is expected that the "kerberos-env" configuration type is available.  It is used to obtain
+   * information about the relevant KDC.  For example, the "kdc_type" property is used to determine
+   * what type of KDC is being used so that the appropriate actions maybe taken in order interact
+   * with it properly.
+   * <p/>
+   * It is expected tht the "kerberos-env" configuration type is available.   It is used to obtain
+   * information about the Kerberos configuration, generally specific to the KDC being used.
+   *
+   * @param cluster                the relevant Cluster
+   * @param serviceComponentFilter a Map of service names to component names indicating the relevant
+   *                               set of services and components - if null, no filter is relevant;
+   *                               if empty, the filter indicates no relevant services or components
+   * @param identityFilter         a Collection of identity names indicating the relevant identities -
+   *                               if null, no filter is relevant; if empty, the filter indicates no
+   *                               relevant identities
+   * @param requestStageContainer  a RequestStageContainer to place generated stages, if needed -
+   *                               if null a new RequestStageContainer will be created.
+   * @return the updated or a new RequestStageContainer containing the stages that need to be
+   * executed to complete this task; or null if no stages need to be executed.
+   * @throws AmbariException
+   * @throws KerberosOperationException
+   * @throws KerberosInvalidConfigurationException if an issue occurs trying to get the
+   *                                               Kerberos-specific configuration details
+   */
+  public RequestStageContainer deleteIdentities(Cluster cluster, Map<String, ? extends Collection<String>> serviceComponentFilter,
+                                                Collection<String> identityFilter, RequestStageContainer requestStageContainer)
+      throws AmbariException, KerberosOperationException {
+    return handle(cluster, getKerberosDetails(cluster), serviceComponentFilter, identityFilter,
+        requestStageContainer, new DeletePrincipalsAndKeytabsHandler());
   }
 
   /**
@@ -487,6 +514,8 @@ public class KerberosHelper {
    * @return the updated or a new RequestStageContainer containing the stages that need to be
    * executed to complete this task; or null if no stages need to be executed.
    * @throws AmbariException
+   * @throws KerberosInvalidConfigurationException if an issue occurs trying to get the
+   * Kerberos-specific configuration details
    */
   @Transactional
   private RequestStageContainer handle(Cluster cluster,
@@ -494,7 +523,7 @@ public class KerberosHelper {
                                        Map<String, ? extends Collection<String>> serviceComponentFilter,
                                        Collection<String> identityFilter,
                                        RequestStageContainer requestStageContainer,
-                                       Handler handler) throws AmbariException {
+                                       Handler handler) throws AmbariException, KerberosOperationException {
 
     Map<String, Service> services = cluster.getServices();
 
@@ -647,7 +676,7 @@ public class KerberosHelper {
                   dataDirectory.getAbsolutePath(), t.getMessage()), t);
             }
 
-            throw new IllegalArgumentException(e.getMessage(), e);
+            throw e;
           }
 
           setAuthToLocalRules(kerberosDescriptor, cluster, kerberosDetails.getDefaultRealm(),
@@ -749,8 +778,6 @@ public class KerberosHelper {
    *
    * @param cluster associated cluster
    *
-   * @throws IllegalArgumentException if the credentials are missing or invalid or
-   *                                  if any associated configuration is invalid
    * @throws AmbariException if any other error occurs while trying to validate the credentials
    */
   public void validateKDCCredentials(Cluster cluster) throws KerberosMissingAdminCredentialsException,
@@ -1450,6 +1477,29 @@ public class KerberosHelper {
     return cluster.getSecurityType() == SecurityType.KERBEROS;
   }
 
+  /**
+   * Tests the request properties to check for directives that need to be acted upon
+   * <p/>
+   * It is required that the SecurityType from the request is either 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
+   * @return true if custom operations should be executed; false otherwise
+   */
+  public boolean shouldExecuteCustomOperations(SecurityType requestSecurityType, Map<String, String> requestProperties) {
+
+    if (((requestSecurityType == SecurityType.KERBEROS) || (requestSecurityType == SecurityType.NONE)) &&
+        (requestProperties != null) && !requestProperties.isEmpty()) {
+      for (SupportedCustomOperation type : SupportedCustomOperation.values()) {
+        if (requestProperties.containsKey(type.name().toLowerCase())) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+
   /**
    * Given a list of KerberosIdentityDescriptors, returns a Map fo configuration types to property
    * names and values.
@@ -1693,6 +1743,40 @@ public class KerberosHelper {
       requestStageContainer.addStages(roleGraph.getStages());
     }
 
+    public void addDeleteKeytabFilesStage(Cluster cluster, List<ServiceComponentHost> serviceComponentHosts,
+                                              String clusterHostInfoJson, String hostParamsJson,
+                                              Map<String, String> commandParameters,
+                                              RoleCommandOrder roleCommandOrder,
+                                              RequestStageContainer requestStageContainer)
+        throws AmbariException {
+      Stage stage = createNewStage(requestStageContainer.getLastStageId(),
+          cluster,
+          requestStageContainer.getId(),
+          "Delete Keytabs",
+          clusterHostInfoJson,
+          StageUtils.getGson().toJson(commandParameters),
+          hostParamsJson);
+
+      if (!serviceComponentHosts.isEmpty()) {
+        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);
+        requestResourceFilters.add(reqResFilter);
+
+        ActionExecutionContext actionExecContext = new ActionExecutionContext(
+            cluster.getClusterName(),
+            "REMOVE_KEYTAB",
+            requestResourceFilters,
+            requestParams);
+        customCommandExecutionHelper.addExecutionCommandsToStage(actionExecContext, stage, requestParams, false);
+      }
+
+      RoleGraph roleGraph = new RoleGraph(roleCommandOrder);
+      roleGraph.build(stage);
+      requestStageContainer.addStages(roleGraph.getStages());
+    }
+
     public void addUpdateConfigurationsStage(Cluster cluster, String clusterHostInfoJson,
                                              String hostParamsJson, ServiceComponentHostServerActionEvent event,
                                              Map<String, String> commandParameters,
@@ -1990,6 +2074,11 @@ public class KerberosHelper {
       addDestroyPrincipalsStage(cluster, clusterHostInfoJson, hostParamsJson, event, commandParameters,
           roleCommandOrder, requestStageContainer);
 
+      // *****************************************************************
+      // Create stage to delete keytabs
+      addDeleteKeytabFilesStage(cluster, serviceComponentHosts, clusterHostInfoJson,
+          hostParamsJson, commandParameters, roleCommandOrder, requestStageContainer);
+
       return requestStageContainer.getLastStageId();
     }
   }
@@ -2095,29 +2184,83 @@ public class KerberosHelper {
   }
 
   /**
-   * Method used to externally peek if weather we have supported operations to execute or not
+   * DeletePrincipalsAndKeytabsHandler is an implementation of the Handler interface used to delete
+   * principals and keytabs throughout the cluster.
    * <p/>
-   * 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
-   * @return true if custom operations should be executed; false otherwise
+   * To complete the process, this implementation creates the following stages:
+   * <ol>
+   * <li>delete principals</li>
+   * <li>remove keytab files</li>
+   * </ol>
    */
-  public boolean shouldExecuteCustomOperations(SecurityType requestSecurityType, Map<String, String> requestProperties) {
 
-    if (((requestSecurityType == SecurityType.KERBEROS) || (requestSecurityType == SecurityType.NONE)) &&
-        (requestProperties != null) && !requestProperties.isEmpty()) {
-      for (SupportedCustomOperation type : SupportedCustomOperation.values()) {
-        if (requestProperties.containsKey(type.name().toLowerCase())) {
-          return true;
-        }
+  private class DeletePrincipalsAndKeytabsHandler extends Handler {
+
+    @Override
+    public boolean shouldProcess(SecurityState desiredSecurityState, ServiceComponentHost sch) throws AmbariException {
+      return true;
+    }
+
+    @Override
+    public SecurityState getNewDesiredSCHSecurityState() {
+      return null;
+    }
+
+    @Override
+    public SecurityState getNewSCHSecurityState() {
+      return null;
+    }
+
+    @Override
+    public SecurityState getNewServiceSecurityState() {
+      return null;
+    }
+
+    @Override
+    public long createStages(Cluster cluster, Map<String, Host> hosts,
+                             Map<String, Map<String, String>> kerberosConfigurations,
+                             String clusterHostInfoJson, String hostParamsJson,
+                             ServiceComponentHostServerActionEvent event,
+                             RoleCommandOrder roleCommandOrder, KerberosDetails kerberosDetails,
+                             File dataDirectory, RequestStageContainer requestStageContainer,
+                             List<ServiceComponentHost> serviceComponentHosts)
+        throws AmbariException {
+      // If there are principals and keytabs to process, setup the following sages:
+      //  1) delete principals
+      //  2) delete keytab files
+
+      // If a RequestStageContainer does not already exist, create a new one...
+      if (requestStageContainer == null) {
+        requestStageContainer = new RequestStageContainer(
+            actionManager.getNextRequestId(),
+            null,
+            requestFactory,
+            actionManager);
       }
+
+      Map<String, String> commandParameters = new HashMap<String, String>();
+      commandParameters.put(KerberosServerAction.AUTHENTICATED_USER_NAME, ambariManagementController.getAuthName());
+      commandParameters.put(KerberosServerAction.DATA_DIRECTORY, dataDirectory.getAbsolutePath());
+      commandParameters.put(KerberosServerAction.DEFAULT_REALM, kerberosDetails.getDefaultRealm());
+      commandParameters.put(KerberosServerAction.KDC_TYPE, kerberosDetails.getKdcType().name());
+      commandParameters.put(KerberosServerAction.ADMINISTRATOR_CREDENTIAL, getEncryptedAdministratorCredentials(cluster));
+
+      // *****************************************************************
+      // Create stage to delete principals
+      addDestroyPrincipalsStage(cluster, clusterHostInfoJson, hostParamsJson, event,
+          commandParameters, roleCommandOrder, requestStageContainer);
+
+      // *****************************************************************
+      // Create stage to delete keytabs
+      addDeleteKeytabFilesStage(cluster, serviceComponentHosts, clusterHostInfoJson,
+          hostParamsJson, commandParameters, roleCommandOrder, requestStageContainer);
+
+      return requestStageContainer.getLastStageId();
     }
-    return false;
   }
 
 
+
   /**
    * KerberosDetails is a helper class to hold the details of the relevant Kerberos-specific
    * configurations so they may be passed around more easily.

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

@@ -178,6 +178,18 @@ public class KerberosPrincipalHostDAO {
     remove(findByHost(hostName));
   }
 
+  /**
+   * Remove KerberosPrincipalHostEntity instance for the specified principal and host
+   *
+   * @param principalName a String indicating the name of the principal
+   * @param hostName      a String indicating the name of the host
+   * @see #remove(org.apache.ambari.server.orm.entities.KerberosPrincipalHostEntity)
+   */
+  @Transactional
+  public void remove(String principalName, String hostName) {
+    remove(new KerberosPrincipalHostEntity(principalName, hostName));
+  }
+
   /**
    * Tests the existence of a principal on at least one host
    *

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

@@ -18,6 +18,8 @@
 
 package org.apache.ambari.server.serveraction.kerberos;
 
+import com.google.inject.Inject;
+import com.google.inject.Injector;
 import com.google.inject.Singleton;
 
 /**
@@ -26,6 +28,9 @@ import com.google.inject.Singleton;
 @Singleton
 public class KerberosOperationHandlerFactory {
 
+  @Inject
+  private Injector injector;
+
   /**
    * Gets the relevant KerberosOperationHandler for some KDCType.
    * <p/>
@@ -43,9 +48,9 @@ public class KerberosOperationHandlerFactory {
 
     switch (kdcType) {
       case MIT_KDC:
-        return new MITKerberosOperationHandler();
+        return injector.getInstance(MITKerberosOperationHandler.class);
       case ACTIVE_DIRECTORY:
-        return new ADKerberosOperationHandler();
+        return injector.getInstance(ADKerberosOperationHandler.class);
       default:
         throw new IllegalArgumentException(String.format("Unexpected kdcType value: %s", kdcType.name()));
     }

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

@@ -19,8 +19,6 @@
 package org.apache.ambari.server.serveraction.kerberos;
 
 import com.google.inject.Inject;
-import com.google.inject.Injector;
-import org.apache.ambari.server.StaticallyInject;
 import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.utils.ShellCommandUtil;
 import org.slf4j.Logger;
@@ -43,7 +41,6 @@ import java.util.regex.Pattern;
  * It is assumed that a MIT Kerberos client is installed and that the kdamin shell command is
  * available
  */
-@StaticallyInject
 public class MITKerberosOperationHandler extends KerberosOperationHandler {
 
   /**
@@ -55,7 +52,7 @@ public class MITKerberosOperationHandler extends KerberosOperationHandler {
   private final static Logger LOG = LoggerFactory.getLogger(MITKerberosOperationHandler.class);
 
   @Inject
-  private static Injector injector;
+  private Configuration configuration;
 
   /**
    * Prepares and creates resources to be used by this KerberosOperationHandler
@@ -94,17 +91,6 @@ public class MITKerberosOperationHandler extends KerberosOperationHandler {
     setOpen(false);
   }
 
-  /**
-   * Statically initialize the Injector
-   * <p/>
-   * This should only be used for unit tests.
-   *
-   * @param injector the Injector to (manually) statically inject
-   */
-  public static void init(Injector injector) {
-    MITKerberosOperationHandler.injector = injector;
-  }
-
   /**
    * Test to see if the specified principal exists in a previously configured MIT KDC
    * <p/>
@@ -244,6 +230,18 @@ public class MITKerberosOperationHandler extends KerberosOperationHandler {
     }
   }
 
+  /**
+   * Initialize this MITKerberosOperationHandler with instances of objects that should normally
+   * be injected.
+   * <p/>
+   * This should only be used for unit tests.
+   *
+   * @param configuration the Configuration to (manually) inject
+   */
+  public void init(Configuration configuration) {
+    this.configuration = configuration;
+  }
+
   /**
    * Retrieves the current key number assigned to the identity identified by the specified principal
    *
@@ -327,8 +325,6 @@ public class MITKerberosOperationHandler extends KerberosOperationHandler {
 
       String pathToCommand = "";
 
-      Configuration configuration = injector.getInstance(Configuration.class);
-
       if (configuration.getServerOsFamily().equals("redhat5")) {
         pathToCommand = "/usr/kerberos/sbin/";
       }

ambari-server/src/main/resources/common-services/KERBEROS/1.10.3-10/metainfo.xml

@@ -52,6 +52,14 @@
                 <timeout>1000</timeout>
               </commandScript>
             </customCommand>
+            <customCommand>
+              <name>REMOVE_KEYTAB</name>
+              <commandScript>
+                <script>scripts/kerberos_client.py</script>
+                <scriptType>PYTHON</scriptType>
+                <timeout>1000</timeout>
+              </commandScript>
+            </customCommand>
           </customCommands>
           <configFiles>
             <configFile>

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

@@ -59,5 +59,8 @@ class KerberosClient(KerberosScript):
   def set_keytab(self, env):
     self.write_keytab_file()
 
+  def remove_keytab(self, env):
+    self.delete_keytab_file()
+
 if __name__ == "__main__":
   KerberosClient().execute()

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

@@ -26,6 +26,7 @@ import tempfile
 
 from resource_management import *
 from utils import get_property_value
+from ambari_commons.os_utils import remove_file
 
 class KerberosScript(Script):
   KRB5_REALM_PROPERTIES = [
@@ -405,3 +406,25 @@ class KerberosScript(Script):
               curr_content['keytabs'][principal.replace("_HOST", params.hostname)] = keytab_file_path
 
               self.put_structured_out(curr_content)
+
+  def delete_keytab_file(self):
+    import params
+
+    if params.kerberos_command_params is not None:
+      for item in params.kerberos_command_params:
+        keytab_file_path = get_property_value(item, 'keytab_file_path')
+        if (keytab_file_path is not None) and (len(keytab_file_path) > 0):
+
+          # Delete the keytab file
+          File(keytab_file_path, action="delete")
+
+          principal = get_property_value(item, 'principal')
+          if principal is not None:
+            curr_content = Script.structuredOut
+
+            if "keytabs" not in curr_content:
+              curr_content['keytabs'] = {}
+
+            curr_content['keytabs'][principal.replace("_HOST", params.hostname)] = '_REMOVED_'
+
+            self.put_structured_out(curr_content)

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

@@ -26,6 +26,9 @@ class KerberosServiceCheck(KerberosScript):
   def service_check(self, env):
     import params
 
+    if params.smoke_test_principal is None:
+      params.smoke_test_principal = params.smoke_user
+
     # First attempt to test using the smoke test user, if data is available
     if ((params.smoke_test_principal is not None) and
           (params.smoke_test_keytab_file is not None) and
@@ -37,20 +40,9 @@ class KerberosServiceCheck(KerberosScript):
       }, user=params.smoke_user)
       test_performed = True
 
-    # Else if a test credentials is specified, try to test using that
-    elif params.test_principal is not None:
-      print "Performing kinit using test user: %s" % params.test_principal
-      code, out = self.test_kinit({
-        'principal': params.test_principal,
-        'keytab_file': params.test_keytab_file,
-        'keytab': params.test_keytab,
-        'password': params.test_password
-      }, user=params.smoke_user)
-      test_performed = True
-
     else:
-      code = 0
-      out = ''
+      code = -1
+      out = 'No principal or keytab found'
       test_performed = False
 
     if test_performed:

ambari-server/src/main/resources/stacks/HDP/2.2.GlusterFS/services/KERBEROS/metainfo.xml

@@ -89,6 +89,14 @@
                 <timeout>1000</timeout>
               </commandScript>
             </customCommand>
+            <customCommand>
+              <name>REMOVE_KEYTAB</name>
+              <commandScript>
+                <script>scripts/kerberos_client.py</script>
+                <scriptType>PYTHON</scriptType>
+                <timeout>1000</timeout>
+              </commandScript>
+            </customCommand>
           </customCommands>
           <configFiles>
             <configFile>

ambari-server/src/main/resources/stacks/HDP/2.2.GlusterFS/services/KERBEROS/package/scripts/kerberos_client.py

@@ -37,5 +37,8 @@ class KerberosClient(KerberosScript):
   def set_keytab(self, env):
     KerberosScript.write_keytab_file()
 
+  def remove_keytab(self, env):
+    self.delete_keytab_file()
+
 if __name__ == "__main__":
   KerberosClient().execute()

ambari-server/src/main/resources/stacks/HDP/2.2.GlusterFS/services/KERBEROS/package/scripts/kerberos_common.py

@@ -26,6 +26,7 @@ import tempfile
 
 from resource_management import *
 from utils import get_property_value
+from ambari_commons.os_utils import remove_file
 
 class KerberosScript(Script):
   KRB5_REALM_PROPERTIES = [
@@ -363,6 +364,26 @@ class KerberosScript(Script):
 
           KerberosScript._set_file_access(file_path, params.keytab_details, params.default_group)
 
+  def delete_keytab_file(self):
+    import params
+
+    if params.kerberos_command_params is not None:
+      for item in params.kerberos_command_params:
+        keytab_file_path = get_property_value(item, 'keytab_file_path')
+        if (keytab_file_path is not None) and (len(keytab_file_path) > 0):
+          print 'Removing keytab file, ', keytab_file_path, "\n"
+          remove_file(keytab_file_path)
+
+          principal = get_property_value(item, 'principal')
+          if principal is not None:
+            curr_content = Script.structuredOut
+
+            if "keytabs" not in curr_content:
+              curr_content['keytabs'] = {}
+
+            curr_content['keytabs'][principal.replace("_HOST", params.hostname)] = '_REMOVED_'
+
+            self.put_structured_out(curr_content)
 
   @staticmethod
   def _set_file_access(file_path, access_details, default_group=None):

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

@@ -2485,34 +2485,64 @@ public class TestHeartbeatHandler {
 
   @Test
   public void testInjectKeytabApplicableHost() throws Exception {
-    List<Map<String, String>> kcp = testInjectKeytab("c6403.ambari.apache.org");
+    List<Map<String, String>> kcp;
+    Map<String, String> properties;
 
+    kcp = testInjectKeytabSetKeytab("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));
+
+    properties = kcp.get(0);
+    Assert.assertNotNull(properties);
+    Assert.assertEquals("c6403.ambari.apache.org", properties.get(KerberosActionDataFile.HOSTNAME));
+    Assert.assertEquals("HDFS", properties.get(KerberosActionDataFile.SERVICE));
+    Assert.assertEquals("DATANODE", properties.get(KerberosActionDataFile.COMPONENT));
+    Assert.assertEquals("dn/_HOST@_REALM", properties.get(KerberosActionDataFile.PRINCIPAL));
+    Assert.assertEquals("hdfs-site/dfs.namenode.kerberos.principal", properties.get(KerberosActionDataFile.PRINCIPAL_CONFIGURATION));
+    Assert.assertEquals("/etc/security/keytabs/dn.service.keytab", properties.get(KerberosActionDataFile.KEYTAB_FILE_PATH));
+    Assert.assertEquals("hdfs", properties.get(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME));
+    Assert.assertEquals("r", properties.get(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS));
+    Assert.assertEquals("hadoop", properties.get(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME));
+    Assert.assertEquals("", properties.get(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS));
+    Assert.assertEquals("hdfs-site/dfs.namenode.keytab.file", properties.get(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION));
 
     Assert.assertEquals(Base64.encodeBase64String("hello".getBytes()), kcp.get(0).get(KerberosServerAction.KEYTAB_CONTENT_BASE64));
 
+
+    kcp = testInjectKeytabRemoveKeytab("c6403.ambari.apache.org");
+
+    Assert.assertNotNull(kcp);
+    Assert.assertEquals(1, kcp.size());
+
+    properties = kcp.get(0);
+    Assert.assertNotNull(properties);
+    Assert.assertEquals("c6403.ambari.apache.org", properties.get(KerberosActionDataFile.HOSTNAME));
+    Assert.assertEquals("HDFS", properties.get(KerberosActionDataFile.SERVICE));
+    Assert.assertEquals("DATANODE", properties.get(KerberosActionDataFile.COMPONENT));
+    Assert.assertEquals("dn/_HOST@_REALM", properties.get(KerberosActionDataFile.PRINCIPAL));
+    Assert.assertFalse(properties.containsKey(KerberosActionDataFile.PRINCIPAL_CONFIGURATION));
+    Assert.assertEquals("/etc/security/keytabs/dn.service.keytab", properties.get(KerberosActionDataFile.KEYTAB_FILE_PATH));
+    Assert.assertFalse(properties.containsKey(KerberosActionDataFile.KEYTAB_FILE_OWNER_NAME));
+    Assert.assertFalse(properties.containsKey(KerberosActionDataFile.KEYTAB_FILE_OWNER_ACCESS));
+    Assert.assertFalse(properties.containsKey(KerberosActionDataFile.KEYTAB_FILE_GROUP_NAME));
+    Assert.assertFalse(properties.containsKey(KerberosActionDataFile.KEYTAB_FILE_GROUP_ACCESS));
+    Assert.assertFalse(properties.containsKey(KerberosActionDataFile.KEYTAB_FILE_CONFIGURATION));
+    Assert.assertFalse(properties.containsKey(KerberosServerAction.KEYTAB_CONTENT_BASE64));
   }
 
   @Test
   public void testInjectKeytabNotApplicableHost() throws Exception {
-    List<Map<String, String>> kcp = testInjectKeytab("c6401.ambari.apache.org");
+    List<Map<String, String>> kcp;
+    kcp = testInjectKeytabSetKeytab("c6401.ambari.apache.org");
+    Assert.assertNotNull(kcp);
+    Assert.assertTrue(kcp.isEmpty());
+
+    kcp = testInjectKeytabRemoveKeytab("c6401.ambari.apache.org");
     Assert.assertNotNull(kcp);
     Assert.assertTrue(kcp.isEmpty());
   }
 
-  private List<Map<String, String>> testInjectKeytab(String targetHost) throws Exception {
+  private List<Map<String, String>> testInjectKeytabSetKeytab(String targetHost) throws Exception {
 
     ExecutionCommand executionCommand = new ExecutionCommand();
 
@@ -2537,7 +2567,38 @@ public class TestHeartbeatHandler {
         }});
     replay(am);
 
-    getHeartBeatHandler(am, aq).injectKeytab(executionCommand, targetHost);
+    getHeartBeatHandler(am, aq).injectKeytab(executionCommand, "SET_KEYTAB", targetHost);
+
+    return executionCommand.getKerberosCommandParams();
+  }
+
+
+  private List<Map<String, String>> testInjectKeytabRemoveKeytab(String targetHost) throws Exception {
+
+    ExecutionCommand executionCommand = new ExecutionCommand();
+
+    Map<String, String> hlp = new HashMap<String, String>();
+    hlp.put("custom_command", "REMOVE_KEYTAB");
+    executionCommand.setHostLevelParams(hlp);
+
+    Map<String, String> commandparams = new HashMap<String, String>();
+    commandparams.put(KerberosServerAction.AUTHENTICATED_USER_NAME, "admin");
+    commandparams.put(KerberosServerAction.DATA_DIRECTORY, createTestKeytabData().getAbsolutePath());
+    executionCommand.setCommandParams(commandparams);
+
+    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, "REMOVE_KEYTAB", targetHost);
 
     return executionCommand.getKerberosCommandParams();
   }

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

@@ -42,6 +42,7 @@ import org.apache.ambari.server.orm.DBAccessor;
 import org.apache.ambari.server.security.SecurityHelper;
 import org.apache.ambari.server.serveraction.kerberos.KDCType;
 import org.apache.ambari.server.serveraction.kerberos.KerberosCredential;
+import org.apache.ambari.server.serveraction.kerberos.KerberosMissingAdminCredentialsException;
 import org.apache.ambari.server.serveraction.kerberos.KerberosOperationException;
 import org.apache.ambari.server.serveraction.kerberos.KerberosOperationHandler;
 import org.apache.ambari.server.serveraction.kerberos.KerberosOperationHandlerFactory;
@@ -90,7 +91,6 @@ import java.util.Map;
 import java.util.Set;
 
 import static org.easymock.EasyMock.*;
-import static org.easymock.EasyMock.createNiceMock;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
@@ -236,7 +236,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     testEnableKerberos(new KerberosCredential("principal", "password", "keytab"), true, false);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test(expected = KerberosMissingAdminCredentialsException.class)
   public void testEnableKerberosMissingCredentials() throws Exception {
     try {
       testEnableKerberos(null, true, false);
@@ -246,7 +246,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     }
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test(expected = KerberosMissingAdminCredentialsException.class)
   public void testEnableKerberosInvalidCredentials() throws Exception {
     try {
       testEnableKerberos(new KerberosCredential("invalid_principal", "password", "keytab"), true, false);
@@ -271,7 +271,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     testEnsureIdentities(new KerberosCredential("principal", "password", "keytab"));
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test(expected = KerberosMissingAdminCredentialsException.class)
   public void testEnsureIdentitiesMissingCredentials() throws Exception {
     try {
       testEnsureIdentities(null);
@@ -281,7 +281,7 @@ public class KerberosHelperTest extends EasyMockSupport {
     }
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test(expected = KerberosMissingAdminCredentialsException.class)
   public void testEnsureIdentitiesInvalidCredentials() throws Exception {
     try {
       testEnsureIdentities(new KerberosCredential("invalid_principal", "password", "keytab"));
@@ -290,6 +290,30 @@ public class KerberosHelperTest extends EasyMockSupport {
       throw e;
     }
   }
+  @Test
+  public void testDeleteIdentities() throws Exception {
+    testDeleteIdentities(new KerberosCredential("principal", "password", "keytab"));
+  }
+
+  @Test(expected = KerberosMissingAdminCredentialsException.class)
+  public void testDeleteIdentitiesMissingCredentials() throws Exception {
+    try {
+      testDeleteIdentities(null);
+    } catch (IllegalArgumentException e) {
+      Assert.assertTrue(e.getMessage().startsWith("Missing KDC administrator credentials"));
+      throw e;
+    }
+  }
+
+  @Test(expected = KerberosMissingAdminCredentialsException.class)
+  public void testDeleteIdentitiesInvalidCredentials() throws Exception {
+    try {
+      testDeleteIdentities(new KerberosCredential("invalid_principal", "password", "keytab"));
+    } catch (IllegalArgumentException e) {
+      Assert.assertTrue(e.getMessage().startsWith("Invalid KDC administrator credentials"));
+      throw e;
+    }
+  }
 
   @Test
   public void testExecuteCustomOperationsInvalidOperation() throws Exception {
@@ -741,6 +765,10 @@ 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"))
@@ -857,7 +885,11 @@ public class KerberosHelperTest extends EasyMockSupport {
     expect(requestStageContainer.getId()).andReturn(1L).once();
     requestStageContainer.addStages(anyObject(List.class));
     expectLastCall().once();
-    // TODO: Add more of these when more stages are added.
+    // Delete Keytabs Stage
+    expect(requestStageContainer.getLastStageId()).andReturn(2L).anyTimes();
+    expect(requestStageContainer.getId()).andReturn(1L).once();
+    requestStageContainer.addStages(anyObject(List.class));
+    expectLastCall().once();
     // Clean-up/Finalize Stage
     expect(requestStageContainer.getLastStageId()).andReturn(3L).anyTimes();
     expect(requestStageContainer.getId()).andReturn(1L).once();
@@ -1162,6 +1194,40 @@ public class KerberosHelperTest extends EasyMockSupport {
     verifyAll();
   }
 
+  @Test
+  public void testIsClusterKerberosEnabled_false() throws Exception {
+    KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
+    Cluster cluster = createStrictMock(Cluster.class);
+
+    expect(cluster.getSecurityType()).andReturn(SecurityType.NONE);
+
+    replay(cluster);
+    assertFalse(kerberosHelper.isClusterKerberosEnabled(cluster));
+    verify(cluster);
+  }
+
+  @Test
+  public void testIsClusterKerberosEnabled_true() throws Exception {
+    KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
+    Cluster cluster = createStrictMock(Cluster.class);
+
+    expect(cluster.getSecurityType()).andReturn(SecurityType.KERBEROS);
+
+    replay(cluster);
+    assertTrue(kerberosHelper.isClusterKerberosEnabled(cluster));
+    verify(cluster);
+  }
+
+  private void setClusterController() throws Exception {
+    KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
+
+    Class<?> c = kerberosHelper.getClass();
+
+    Field f = c.getDeclaredField("clusterController");
+    f.setAccessible(true);
+    f.set(kerberosHelper, clusterController);
+  }
+
   private void setupGetDescriptorFromCluster(KerberosDescriptor kerberosDescriptor) throws Exception {
     ResourceProvider resourceProvider = createStrictMock(ResourceProvider.class);
     expect(clusterController.ensureResourceProvider(Resource.Type.Artifact)).andReturn(resourceProvider).once();
@@ -1495,37 +1561,278 @@ public class KerberosHelperTest extends EasyMockSupport {
     verifyAll();
   }
 
-  @Test
-  public void testIsClusterKerberosEnabled_false() throws Exception {
+  private void testDeleteIdentities(final KerberosCredential kerberosCredential) throws Exception {
     KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
-    Cluster cluster = createStrictMock(Cluster.class);
 
-    expect(cluster.getSecurityType()).andReturn(SecurityType.NONE);
+    final ServiceComponentHost schKerberosClient = createMock(ServiceComponentHost.class);
+    expect(schKerberosClient.getServiceName()).andReturn(Service.Type.KERBEROS.name()).anyTimes();
+    expect(schKerberosClient.getServiceComponentName()).andReturn(Role.KERBEROS_CLIENT.name()).anyTimes();
+    expect(schKerberosClient.getHostName()).andReturn("host1").anyTimes();
+    expect(schKerberosClient.getState()).andReturn(State.INSTALLED).anyTimes();
 
-    replay(cluster);
-    assertFalse(kerberosHelper.isClusterKerberosEnabled(cluster));
-    verify(cluster);
-  }
+    final ServiceComponentHost sch1 = createMock(ServiceComponentHost.class);
+    expect(sch1.getServiceName()).andReturn("SERVICE1").anyTimes();
+    expect(sch1.getServiceComponentName()).andReturn("COMPONENT1").anyTimes();
+    expect(sch1.getHostName()).andReturn("host1").anyTimes();
 
-  @Test
-  public void testIsClusterKerberosEnabled_true() throws Exception {
-    KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
-    Cluster cluster = createStrictMock(Cluster.class);
+    final ServiceComponentHost sch2 = createStrictMock(ServiceComponentHost.class);
+    expect(sch2.getServiceName()).andReturn("SERVICE2").anyTimes();
+    expect(sch2.getServiceComponentName()).andReturn("COMPONENT3").anyTimes();
 
-    expect(cluster.getSecurityType()).andReturn(SecurityType.KERBEROS);
+    final ServiceComponentHost sch3 = createStrictMock(ServiceComponentHost.class);
+    expect(sch3.getServiceName()).andReturn("SERVICE3").anyTimes();
+    expect(sch3.getServiceComponentName()).andReturn("COMPONENT3").anyTimes();
+    expect(sch3.getHostName()).andReturn("host1").anyTimes();
 
-    replay(cluster);
-    assertTrue(kerberosHelper.isClusterKerberosEnabled(cluster));
-    verify(cluster);
-  }
+    final Host host = createNiceMock(Host.class);
+    expect(host.getHostName()).andReturn("host1").anyTimes();
+    expect(host.getState()).andReturn(HostState.HEALTHY).anyTimes();
 
-  private void setClusterController() throws Exception {
-    KerberosHelper kerberosHelper = injector.getInstance(KerberosHelper.class);
+    final ServiceComponent serviceComponentKerberosClient = createNiceMock(ServiceComponent.class);
+    expect(serviceComponentKerberosClient.getName()).andReturn(Role.KERBEROS_CLIENT.name()).anyTimes();
+    expect(serviceComponentKerberosClient.getServiceComponentHosts()).andReturn(Collections.singletonMap("host1", schKerberosClient)).anyTimes();
 
-    Class<?> c = kerberosHelper.getClass();
+    final Service serviceKerberos = createStrictMock(Service.class);
+    expect(serviceKerberos.getName()).andReturn(Service.Type.KERBEROS.name()).anyTimes();
+    expect(serviceKerberos.getServiceComponents())
+        .andReturn(Collections.singletonMap(Role.KERBEROS_CLIENT.name(), serviceComponentKerberosClient))
+        .times(3);
 
-    Field f = c.getDeclaredField("clusterController");
-    f.setAccessible(true);
-    f.set(kerberosHelper, clusterController);
+    final Service service1 = createStrictMock(Service.class);
+    expect(service1.getName()).andReturn("SERVICE1").anyTimes();
+    expect(service1.getServiceComponents())
+        .andReturn(Collections.<String, ServiceComponent>emptyMap())
+        .times(3);
+
+    final Service service2 = createStrictMock(Service.class);
+    expect(service2.getName()).andReturn("SERVICE2").anyTimes();
+    expect(service2.getServiceComponents())
+        .andReturn(Collections.<String, ServiceComponent>emptyMap())
+        .times(3);
+
+    final Map<String, String> kerberosEnvProperties = createNiceMock(Map.class);
+    expect(kerberosEnvProperties.get("kdc_type")).andReturn("mit-kdc").anyTimes();
+    expect(kerberosEnvProperties.get("realm")).andReturn("FOOBAR.COM").anyTimes();
+
+    final Config kerberosEnvConfig = createNiceMock(Config.class);
+    expect(kerberosEnvConfig.getProperties()).andReturn(kerberosEnvProperties).anyTimes();
+
+    final Map<String, String> krb5ConfProperties = createNiceMock(Map.class);
+
+    final Config krb5ConfConfig = createNiceMock(Config.class);
+    expect(krb5ConfConfig.getProperties()).andReturn(krb5ConfProperties).anyTimes();
+
+    final Cluster cluster = createNiceMock(Cluster.class);
+    expect(cluster.getDesiredConfigByType("krb5-conf")).andReturn(krb5ConfConfig).anyTimes();
+    expect(cluster.getDesiredConfigByType("kerberos-env")).andReturn(kerberosEnvConfig).anyTimes();
+    expect(cluster.getClusterName()).andReturn("c1").anyTimes();
+    expect(cluster.getServices())
+        .andReturn(new HashMap<String, Service>() {
+          {
+            put(Service.Type.KERBEROS.name(), serviceKerberos);
+            put("SERVICE1", service1);
+            put("SERVICE2", service2);
+          }
+        })
+        .anyTimes();
+    expect(cluster.getServiceComponentHosts("host1"))
+        .andReturn(new ArrayList<ServiceComponentHost>() {
+          {
+            add(sch1);
+            add(sch2);
+            add(sch3);
+            add(schKerberosClient);
+          }
+        })
+        .once();
+    expect(cluster.getCurrentStackVersion())
+        .andReturn(new StackId("HDP", "2.2"))
+        .anyTimes();
+    expect(cluster.getSessionAttributes()).andReturn(new HashMap<String, Object>() {{
+      if (kerberosCredential != null) {
+        put("kerberos_admin/" + KerberosCredential.KEY_NAME_PRINCIPAL, kerberosCredential.getPrincipal());
+        put("kerberos_admin/" + KerberosCredential.KEY_NAME_PASSWORD, kerberosCredential.getPassword());
+        put("kerberos_admin/" + KerberosCredential.KEY_NAME_KEYTAB, kerberosCredential.getKeytab());
+      }
+    }}).anyTimes();
+
+    final Clusters clusters = injector.getInstance(Clusters.class);
+    expect(clusters.getHostsForCluster("c1"))
+        .andReturn(new HashMap<String, Host>() {
+          {
+            put("host1", host);
+          }
+        })
+        .once();
+    expect(clusters.getHost("host1"))
+        .andReturn(host)
+        .once();
+
+    final AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class);
+    expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, "host1"))
+        .andReturn(Collections.<String, Map<String, String>>emptyMap())
+        .once();
+    expect(ambariManagementController.findConfigurationTagsWithOverrides(cluster, null))
+        .andReturn(Collections.<String, Map<String, String>>emptyMap())
+        .once();
+    expect(ambariManagementController.getRoleCommandOrder(cluster))
+        .andReturn(createNiceMock(RoleCommandOrder.class))
+        .once();
+
+    final ConfigHelper configHelper = injector.getInstance(ConfigHelper.class);
+    expect(configHelper.getEffectiveConfigProperties(anyObject(Cluster.class), anyObject(Map.class)))
+        .andReturn(new HashMap<String, Map<String, String>>() {
+          {
+            put("cluster-env", new HashMap<String, String>() {{
+              put("kerberos_domain", "FOOBAR.COM");
+            }});
+          }
+        })
+        .times(2);
+
+    final KerberosPrincipalDescriptor principalDescriptor1a = createMock(KerberosPrincipalDescriptor.class);
+    expect(principalDescriptor1a.getValue()).andReturn("component1a/_HOST@${realm}").anyTimes();
+    expect(principalDescriptor1a.getType()).andReturn(KerberosPrincipalType.SERVICE).anyTimes();
+    expect(principalDescriptor1a.getLocalUsername()).andReturn(null).anyTimes();
+    expect(principalDescriptor1a.getConfiguration()).andReturn("service1b-site/component1.kerberos.principal").anyTimes();
+
+    final KerberosPrincipalDescriptor principalDescriptor1b = createMock(KerberosPrincipalDescriptor.class);
+    expect(principalDescriptor1b.getValue()).andReturn("component1b/_HOST@${realm}").anyTimes();
+    expect(principalDescriptor1b.getType()).andReturn(KerberosPrincipalType.SERVICE).anyTimes();
+    expect(principalDescriptor1b.getLocalUsername()).andReturn(null).anyTimes();
+    expect(principalDescriptor1b.getConfiguration()).andReturn("service1b-site/component1.kerberos.principal").anyTimes();
+
+    final KerberosPrincipalDescriptor principalDescriptor3 = createMock(KerberosPrincipalDescriptor.class);
+    expect(principalDescriptor3.getValue()).andReturn("component3/${host}@${realm}").anyTimes();
+    expect(principalDescriptor3.getType()).andReturn(KerberosPrincipalType.SERVICE).anyTimes();
+    expect(principalDescriptor3.getLocalUsername()).andReturn(null).anyTimes();
+    expect(principalDescriptor3.getConfiguration()).andReturn("service3-site/component3.kerberos.principal").anyTimes();
+
+    final KerberosKeytabDescriptor keytabDescriptor1 = createMock(KerberosKeytabDescriptor.class);
+    expect(keytabDescriptor1.getFile()).andReturn("${keytab_dir}/service1.keytab").once();
+    expect(keytabDescriptor1.getOwnerName()).andReturn("service1").once();
+    expect(keytabDescriptor1.getOwnerAccess()).andReturn("rw").once();
+    expect(keytabDescriptor1.getGroupName()).andReturn("hadoop").once();
+    expect(keytabDescriptor1.getGroupAccess()).andReturn("").once();
+    expect(keytabDescriptor1.getConfiguration()).andReturn("service1-site/component1.keytab.file").once();
+
+    final KerberosKeytabDescriptor keytabDescriptor3 = createMock(KerberosKeytabDescriptor.class);
+    expect(keytabDescriptor3.getFile()).andReturn("${keytab_dir}/service3.keytab").once();
+    expect(keytabDescriptor3.getOwnerName()).andReturn("service3").once();
+    expect(keytabDescriptor3.getOwnerAccess()).andReturn("rw").once();
+    expect(keytabDescriptor3.getGroupName()).andReturn("hadoop").once();
+    expect(keytabDescriptor3.getGroupAccess()).andReturn("").once();
+    expect(keytabDescriptor3.getConfiguration()).andReturn("service3-site/component3.keytab.file").once();
+
+    final KerberosIdentityDescriptor identityDescriptor1a = createMock(KerberosIdentityDescriptor.class);
+    expect(identityDescriptor1a.getName()).andReturn("identity1a").anyTimes();
+    expect(identityDescriptor1a.getPrincipalDescriptor()).andReturn(principalDescriptor1a).anyTimes();
+    expect(identityDescriptor1a.getKeytabDescriptor()).andReturn(keytabDescriptor1).anyTimes();
+
+    final KerberosIdentityDescriptor identityDescriptor1b = createMock(KerberosIdentityDescriptor.class);
+    expect(identityDescriptor1b.getName()).andReturn("identity1b").anyTimes();
+    expect(identityDescriptor1b.getPrincipalDescriptor()).andReturn(principalDescriptor1b).anyTimes();
+
+    final KerberosIdentityDescriptor identityDescriptor3 = createMock(KerberosIdentityDescriptor.class);
+    expect(identityDescriptor3.getName()).andReturn("identity3").anyTimes();
+    expect(identityDescriptor3.getPrincipalDescriptor()).andReturn(principalDescriptor3).anyTimes();
+    expect(identityDescriptor3.getKeytabDescriptor()).andReturn(keytabDescriptor3).anyTimes();
+
+    final ArrayList<KerberosIdentityDescriptor> identityDescriptors1 = new ArrayList<KerberosIdentityDescriptor>() {{
+      add(identityDescriptor1a);
+      add(identityDescriptor1b);
+    }};
+    final KerberosComponentDescriptor componentDescriptor1 = createStrictMock(KerberosComponentDescriptor.class);
+    expect(componentDescriptor1.getIdentities(true)).andReturn(identityDescriptors1).times(1);
+    expect(componentDescriptor1.getConfigurations(true)).andReturn(null).times(1);
+    expect(componentDescriptor1.getIdentities(true)).andReturn(identityDescriptors1).times(1);
+    expect(componentDescriptor1.getAuthToLocalProperties()).andReturn(null).times(1);
+
+    final ArrayList<KerberosIdentityDescriptor> identityDescriptors3 = new ArrayList<KerberosIdentityDescriptor>() {{
+      add(identityDescriptor3);
+    }};
+    final KerberosComponentDescriptor componentDescriptor3 = createStrictMock(KerberosComponentDescriptor.class);
+    expect(componentDescriptor3.getIdentities(true)).andReturn(identityDescriptors3).times(1);
+    expect(componentDescriptor3.getConfigurations(true)).andReturn(null).times(1);
+
+    final KerberosServiceDescriptor serviceDescriptor1 = createMock(KerberosServiceDescriptor.class);
+    expect(serviceDescriptor1.getIdentities(true)).andReturn(null).times(1);
+    expect(serviceDescriptor1.getName()).andReturn("SERVICE1").times(1);
+    expect(serviceDescriptor1.getIdentities(true)).andReturn(null).times(1);
+    expect(serviceDescriptor1.getComponents()).andReturn(new HashMap<String, KerberosComponentDescriptor>(){{
+      put("COMPONENT1", componentDescriptor1);
+    }}).times(1);
+    expect(serviceDescriptor1.getComponent("COMPONENT1")).andReturn(componentDescriptor1).once();
+    expect(serviceDescriptor1.getAuthToLocalProperties()).andReturn(null).once();
+
+    final KerberosServiceDescriptor serviceDescriptor3 = createMock(KerberosServiceDescriptor.class);
+    expect(serviceDescriptor3.getIdentities(true)).andReturn(null).times(1);
+    expect(serviceDescriptor3.getName()).andReturn("SERVICE3").times(1);
+    expect(serviceDescriptor3.getComponent("COMPONENT3")).andReturn(componentDescriptor3).once();
+
+    final KerberosDescriptor kerberosDescriptor = createStrictMock(KerberosDescriptor.class);
+    expect(kerberosDescriptor.getProperties()).andReturn(null).once();
+    expect(kerberosDescriptor.getService("SERVICE1")).andReturn(serviceDescriptor1).once();
+    expect(kerberosDescriptor.getService("SERVICE3")).andReturn(serviceDescriptor3).once();
+    expect(kerberosDescriptor.getIdentities()).andReturn(null).once();
+    expect(kerberosDescriptor.getAuthToLocalProperties()).andReturn(null).once();
+    expect(kerberosDescriptor.getServices()).andReturn(new HashMap<String, KerberosServiceDescriptor>()
+    {{
+      put("SERVCE1", serviceDescriptor1);
+      put("SERVCE2", serviceDescriptor3);
+    }}).once();
+
+    setupGetDescriptorFromCluster(kerberosDescriptor);
+
+    final StageFactory stageFactory = injector.getInstance(StageFactory.class);
+    expect(stageFactory.createNew(anyLong(), anyObject(String.class), anyObject(String.class),
+        anyLong(), anyObject(String.class), anyObject(String.class), anyObject(String.class),
+        anyObject(String.class)))
+        .andAnswer(new IAnswer<Stage>() {
+          @Override
+          public Stage answer() throws Throwable {
+            Stage stage = createNiceMock(Stage.class);
+
+            expect(stage.getHostRoleCommands())
+                .andReturn(Collections.<String, Map<String, HostRoleCommand>>emptyMap())
+                .anyTimes();
+            replay(stage);
+            return stage;
+          }
+        })
+        .anyTimes();
+
+    // This is a STRICT mock to help ensure that the end result is what we want.
+    final RequestStageContainer requestStageContainer = createStrictMock(RequestStageContainer.class);
+    // Delete Principals Stage
+    expect(requestStageContainer.getLastStageId()).andReturn(-1L).anyTimes();
+    expect(requestStageContainer.getId()).andReturn(1L).once();
+    requestStageContainer.addStages(anyObject(List.class));
+    expectLastCall().once();
+    // Delete Keytabs Stage
+    expect(requestStageContainer.getLastStageId()).andReturn(-1L).anyTimes();
+    expect(requestStageContainer.getId()).andReturn(1L).once();
+    requestStageContainer.addStages(anyObject(List.class));
+    expectLastCall().once();
+    // Clean-up/Finalize Stage
+    expect(requestStageContainer.getLastStageId()).andReturn(-1L).anyTimes();
+    expect(requestStageContainer.getId()).andReturn(1L).once();
+    requestStageContainer.addStages(anyObject(List.class));
+    expectLastCall().once();
+
+    replayAll();
+
+    // Needed by infrastructure
+    injector.getInstance(AmbariMetaInfo.class).init();
+
+    Map<String, Collection<String>> serviceComponentFilter = new HashMap<String, Collection<String>>();
+    Collection<String> identityFilter = Arrays.asList("identity1a", "identity3");
+
+    serviceComponentFilter.put("SERVICE3", Collections.singleton("COMPONENT3"));
+    serviceComponentFilter.put("SERVICE1", null);
+
+    kerberosHelper.deleteIdentities(cluster, serviceComponentFilter, identityFilter, requestStageContainer);
+
+    verifyAll();
   }
 }

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

@@ -18,22 +18,49 @@
 
 package org.apache.ambari.server.serveraction.kerberos;
 
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.state.Clusters;
+import org.easymock.EasyMock;
 import org.junit.Assert;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+
 public class KerberosOperationHandlerFactoryTest {
 
 
+  private static Injector injector;
+
+  @BeforeClass
+  public static void beforeClass() throws AmbariException {
+    injector = Guice.createInjector(new AbstractModule() {
+      @Override
+      protected void configure() {
+        Configuration configuration = EasyMock.createNiceMock(Configuration.class);
+        expect(configuration.getServerOsFamily()).andReturn("redhat6").anyTimes();
+        replay(configuration);
+
+        bind(Configuration.class).toInstance(configuration);
+      }
+    });
+  }
+
   @Test
   public void testForAD() {
     Assert.assertEquals(MITKerberosOperationHandler.class,
-      new KerberosOperationHandlerFactory().getKerberosOperationHandler(KDCType.MIT_KDC).getClass());
+      injector.getInstance(KerberosOperationHandlerFactory.class).getKerberosOperationHandler(KDCType.MIT_KDC).getClass());
   }
 
   @Test
   public void testForMIT() {
     Assert.assertEquals(ADKerberosOperationHandler.class,
-      new KerberosOperationHandlerFactory().getKerberosOperationHandler(KDCType.ACTIVE_DIRECTORY).getClass());
+        injector.getInstance(KerberosOperationHandlerFactory.class).getKerberosOperationHandler(KDCType.ACTIVE_DIRECTORY).getClass());
   }
 
   @Test(expected = IllegalArgumentException.class)

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

@@ -23,21 +23,23 @@ import com.google.inject.Guice;
 import com.google.inject.Injector;
 import junit.framework.Assert;
 import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.controller.KerberosHelper;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.utils.ShellCommandUtil;
+import org.easymock.EasyMock;
 import org.easymock.IAnswer;
 import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;
 
+import java.lang.reflect.Field;
 import java.util.HashMap;
 import java.util.Map;
 
 import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.replay;
-import static org.mockito.Mockito.mock;
-
 
 public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTest {
 
@@ -47,21 +49,30 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
 
   private static Injector injector;
 
-  @BeforeClass
-  public static void beforeClass() throws AmbariException {
-    injector = Guice.createInjector(new MockModule());
-    MITKerberosOperationHandler.init(injector);
-  }
-
   private static final Map<String, String> KERBEROS_ENV_MAP = new HashMap<String, String>() {
     {
       put(MITKerberosOperationHandler.KERBEROS_ENV_ENCRYPTION_TYPES, null);
     }
   };
 
+  @BeforeClass
+  public static void beforeClass() throws AmbariException {
+    injector = Guice.createInjector(new AbstractModule() {
+      @Override
+      protected void configure() {
+        Configuration configuration =EasyMock.createNiceMock(Configuration.class);
+        expect(configuration.getServerOsFamily()).andReturn("redhat6").anyTimes();
+        replay(configuration);
+
+        bind(Clusters.class).toInstance(EasyMock.createNiceMock(Clusters.class));
+        bind(Configuration.class).toInstance(configuration);
+      }
+    });
+  }
+
   @Test
   public void testSetPrincipalPasswordExceptions() throws Exception {
-    MITKerberosOperationHandler handler = new MITKerberosOperationHandler();
+    MITKerberosOperationHandler handler = injector.getInstance(MITKerberosOperationHandler.class);
     handler.open(new KerberosCredential(DEFAULT_ADMIN_PRINCIPAL, DEFAULT_ADMIN_PASSWORD, null), DEFAULT_REALM, KERBEROS_ENV_MAP);
 
     try {
@@ -135,6 +146,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -168,6 +181,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -201,6 +216,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -234,6 +251,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -267,6 +286,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -300,6 +321,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -333,6 +356,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -366,6 +391,8 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
         .addMockedMethod(KerberosOperationHandler.class.getDeclaredMethod("executeCommand", String[].class))
         .createNiceMock();
 
+    setConfiguration(handler, "redhat6");
+
     expect(handler.executeCommand(anyObject(String[].class)))
         .andAnswer(new IAnswer<ShellCommandUtil.Result>() {
           @Override
@@ -440,11 +467,11 @@ public class MITKerberosOperationHandlerTest extends KerberosOperationHandlerTes
     handler.close();
   }
 
-  public static class MockModule extends AbstractModule {
-    @Override
-    protected void configure() {
-      bind(Clusters.class).toInstance(mock(Clusters.class));
-    }
-  }
+  private static void setConfiguration(MITKerberosOperationHandler handler, String osType) throws Exception {
+    Configuration configuration = EasyMock.createNiceMock(Configuration.class);
+    expect(configuration.getServerOsFamily()).andReturn("redhat6").anyTimes();
+    replay(configuration);
 
+    handler.init(configuration);
+  }
 }

ambari-server/src/test/python/stacks/2.2/KERBEROS/test_kerberos_client.py

@@ -311,3 +311,39 @@ class TestKerberosClient(RMFTestCase):
                                                    "PK6UkwyUSMAAAA3AAEAC0VYQU1QTEUuQ09NAAlhbWJhcmkt"
                                                    "cWEAAAABVKHYCgEAEQAQVqISRJwXIQnG28lI34mfeA=="))
     )
+
+  def test_delete_keytab(self):
+    config_file = "stacks/2.2/configs/default.json"
+
+    with open(config_file, "r") as f:
+      json_data = json.load(f)
+
+    json_data['kerberosCommandParams'] = []
+    json_data['kerberosCommandParams'].append({
+      "service": "HDFS",
+      "hostname": "c6501.ambari.apache.org",
+      "component": "NAMENODE",
+      "keytab_file_path": "/etc/security/keytabs/spnego.service.keytab",
+      "principal": "HTTP/_HOST@EXAMPLE.COM"
+    })
+
+    json_data['kerberosCommandParams'].append({
+      "service": "HDFS",
+      "hostname": "c6501.ambari.apache.org",
+      "component": "NAMENODE",
+      "keytab_file_path": "/etc/security/keytabs/smokeuser.headless.keytab",
+      "principal": "ambari-qa@EXAMPLE.COM"
+    })
+
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/kerberos_client.py",
+                       classname="KerberosClient",
+                       command="remove_keytab",
+                       config_dict=json_data,
+                       hdp_stack_version=self.STACK_VERSION,
+                       target=RMFTestCase.TARGET_COMMON_SERVICES
+    )
+
+    self.assertResourceCalled('File', "/etc/security/keytabs/spnego.service.keytab",
+                              action=['delete'])
+    self.assertResourceCalled('File', "/etc/security/keytabs/smokeuser.headless.keytab",
+                              action=['delete'])