private RequestStageContainer handleTestIdentity(Cluster cluster,
                                                   KerberosDetails kerberosDetails,
                                                   Map<String, String> commandParameters, RequestStageContainer requestStageContainer,
                                                   Handler handler) throws AmbariException, KerberosOperationException {

    if (kerberosDetails.manageIdentities()) {
      if (commandParameters == null) {
        throw new AmbariException("The properties map must not be null.  It is needed to store data related to the service check identity");
      }

      List<ServiceComponentHost> serviceComponentHostsToProcess = new ArrayList<>();
      KerberosDescriptor kerberosDescriptor = getKerberosDescriptor(cluster, false);

      // This is needed to help determine which hosts to perform actions for and create tasks for.
      Set<String> hostsWithValidKerberosClient = getHostsWithValidKerberosClient(cluster);

      // Create a temporary directory to store metadata needed to complete this task.  Information
      // such as which principals and keytabs files to create as well as what configurations need
      // to be update are stored in data files in this directory. Any keytab files are stored in
      // this directory until they are distributed to their appropriate hosts.
      File dataDirectory = createTemporaryDirectory();

      // Calculate the current non-host-specific configurations. These will be used to replace
      // variables within the Kerberos descriptor data
      Map<String, Map<String, String>> configurations = calculateConfigurations(cluster, null,
        kerberosDescriptor, false, false, null);

      String principal = variableReplacementHelper.replaceVariables("${kerberos-env/service_check_principal_name}@${realm}", configurations);

      String keytabFilePath = variableReplacementHelper.replaceVariables("${keytab_dir}/kerberos.service_check.${short_date}.keytab", configurations);
      String keytabFileOwnerName = variableReplacementHelper.replaceVariables("${cluster-env/smokeuser}", configurations);
      String keytabFileOwnerAccess = "rw";
      String keytabFileGroupName = variableReplacementHelper.replaceVariables("${cluster-env/user_group}", configurations);
      String keytabFileGroupAccess = "r";

      // Add the relevant principal name and keytab file data to the command params state
      commandParameters.put("principal_name", principal);
      commandParameters.put("keytab_file", keytabFilePath);

      try {
        // Get a list KERBEROS/KERBEROS_CLIENT ServiceComponentHost objects
        List<ServiceComponentHost> serviceComponentHosts = cluster.getServiceComponentHosts(Service.Type.KERBEROS.name(), Role.KERBEROS_CLIENT.name());

        if ((serviceComponentHosts != null) && !serviceComponentHosts.isEmpty()) {

          // Iterate over the KERBEROS_CLIENT service component hosts to get the service and
          // component-level Kerberos descriptors in order to determine which principals,
          // keytab files needed to be created or updated.
          for (ServiceComponentHost sch : serviceComponentHosts) {
            if (sch.getState() == State.INSTALLED) {
              String hostname = sch.getHostName();
              KerberosKeytabEntity kke = kerberosKeytabDAO.find(keytabFilePath);

              if (kke == null) {
                kke = new KerberosKeytabEntity();
                kke.setKeytabPath(keytabFilePath);
                kke.setOwnerName(keytabFileOwnerName);
                kke.setOwnerAccess(keytabFileOwnerAccess);
                kke.setGroupName(keytabFileGroupName);
                kke.setGroupAccess(keytabFileGroupAccess);
                kerberosKeytabDAO.create(kke);
              }

              // create principals
              KerberosPrincipalEntity kpe = kerberosPrincipalDAO.find(principal);
              if (kpe == null) {
                kpe = new KerberosPrincipalEntity(principal, false, null);
                kerberosPrincipalDAO.create(kpe);
              }

              KeytabPrincipalFindOrCreateResult result = kerberosKeytabPrincipalDAO.findOrCreate(
                  kke, hostDAO.findById(sch.getHost().getHostId()), kpe, null);

              KerberosKeytabPrincipalEntity kkp = result.kkp;
              if (kkp.putServiceMapping(sch.getServiceName(), sch.getServiceComponentName())) {
                kerberosKeytabPrincipalDAO.merge(kkp);
              }
              kerberosKeytabDAO.merge(kke);
              kerberosPrincipalDAO.merge(kpe);
              hostsWithValidKerberosClient.add(hostname);
              serviceComponentHostsToProcess.add(sch);
            }
          }
        }

      } catch (Exception e) {
        // make sure to log what is going wrong
        LOG.error("Failed " + e);
        throw e;
      }

      // If there are ServiceComponentHosts to process, make sure the administrator credential
      // are available
      if (!serviceComponentHostsToProcess.isEmpty()) {
        try {
          validateKDCCredentials(kerberosDetails, cluster);
        } catch (Exception e) {
          LOG.error("Cannot validate credentials: " + e);
          try {
            FileUtils.deleteDirectory(dataDirectory);
          } catch (Throwable t) {
            LOG.warn(String.format("The data directory (%s) was not deleted due to an error condition - {%s}",
              dataDirectory.getAbsolutePath(), t.getMessage()), t);
          }

          throw e;
        }
      }

      // Always set up the necessary stages to perform the tasks needed to complete the operation.
      // Some stages may be no-ops, this is expected.
      // Gather data needed to create stages and tasks...
      Map<String, Set<String>> clusterHostInfo = StageUtils.getClusterHostInfo(cluster);
      String clusterHostInfoJson = StageUtils.getGson().toJson(clusterHostInfo);

      @Experimental(feature = ExperimentalFeature.MULTI_SERVICE, comment = "The cluster stack id is deprecated")
      Map<String, String> hostParams = customCommandExecutionHelper.createDefaultHostParams(cluster, cluster.getDesiredStackVersion());
      String hostParamsJson = StageUtils.getGson().toJson(hostParams);
      String ambariServerHostname = StageUtils.getHostName();
      ServiceComponentHostServerActionEvent event = new ServiceComponentHostServerActionEvent(
        RootComponent.AMBARI_SERVER.name(),
        ambariServerHostname, // TODO: Choose a random hostname from the cluster. All tasks for the AMBARI_SERVER service will be executed on this Ambari server
        System.currentTimeMillis());
      RoleCommandOrder roleCommandOrder = ambariManagementController.getRoleCommandOrder(cluster);
      // If a RequestStageContainer does not already exist, create a new one...
      if (requestStageContainer == null) {
        requestStageContainer = new RequestStageContainer(
          actionManager.getNextRequestId(),
          null,
          requestFactory,
          actionManager);
      }

      // Use the handler implementation to setup the relevant stages.
      // Set the service/component filter to an empty map since the service/component processing
      // was done above.
      handler.createStages(cluster,
        clusterHostInfoJson, hostParamsJson, event, roleCommandOrder, kerberosDetails,
        dataDirectory, requestStageContainer, serviceComponentHostsToProcess,
        null, null, Sets.newHashSet(principal), hostsWithValidKerberosClient);


      handler.addFinalizeOperationStage(cluster, clusterHostInfoJson, hostParamsJson, event,
        dataDirectory, roleCommandOrder, requestStageContainer, kerberosDetails);
    }

    return requestStageContainer;
  }
Performs operations needed to process Kerberos related tasks to manage a (unique) test identity
on the relevant cluster.
<p/>
If Ambari is not managing Kerberos identities, than this method does nothing.

@param cluster               the relevant Cluster
@param kerberosDetails       a KerberosDetails containing information about relevant Kerberos
                             configuration
@param commandParameters     the command parameters map used to read and/or write attributes
                             related to this operation
@param requestStageContainer a RequestStageContainer to place generated stages, if needed -
                             if null a new RequestStageContainer will be created.
@param handler               a Handler to use to provide guidance and set up stages
                             to perform the work needed to complete the relative action
@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
ambari-server/src/main/java/org/apache/ambari/server/controller/KerberosHelperImpl.java