/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.yarn.service.client;

import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.registry.client.api.RegistryConstants;
import org.apache.hadoop.registry.client.api.RegistryOperations;
import org.apache.hadoop.registry.client.api.RegistryOperationsFactory;
import org.apache.hadoop.registry.client.binding.RegistryUtils;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.VersionInfo;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest;

import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.AppAdminClient;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.client.cli.ApplicationCLI;
import org.apache.hadoop.yarn.client.util.YarnClientUtils;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.proto.ClientAMProtocol.ComponentCountProto;
import org.apache.hadoop.yarn.proto.ClientAMProtocol.FlexComponentsRequestProto;
import org.apache.hadoop.yarn.proto.ClientAMProtocol.GetStatusRequestProto;
import org.apache.hadoop.yarn.proto.ClientAMProtocol.GetStatusResponseProto;
import org.apache.hadoop.yarn.proto.ClientAMProtocol.RestartServiceRequestProto;
import org.apache.hadoop.yarn.proto.ClientAMProtocol.StopRequestProto;
import org.apache.hadoop.yarn.proto.ClientAMProtocol.UpgradeServiceRequestProto;
import org.apache.hadoop.yarn.service.ClientAMProtocol;
import org.apache.hadoop.yarn.service.ServiceMaster;
import org.apache.hadoop.yarn.service.api.records.Component;
import org.apache.hadoop.yarn.service.api.records.Service;
import org.apache.hadoop.yarn.service.api.records.ServiceState;
import org.apache.hadoop.yarn.service.conf.SliderExitCodes;
import org.apache.hadoop.yarn.service.conf.YarnServiceConf;
import org.apache.hadoop.yarn.service.conf.YarnServiceConstants;
import org.apache.hadoop.yarn.service.containerlaunch.ClasspathConstructor;
import org.apache.hadoop.yarn.service.containerlaunch.JavaCommandLineBuilder;
import org.apache.hadoop.yarn.service.exceptions.BadClusterStateException;
import org.apache.hadoop.yarn.service.exceptions.BadConfigException;
import org.apache.hadoop.yarn.service.exceptions.SliderException;
import org.apache.hadoop.yarn.service.provider.AbstractClientProvider;
import org.apache.hadoop.yarn.service.provider.ProviderUtils;
import org.apache.hadoop.yarn.service.utils.ServiceApiUtil;
import org.apache.hadoop.yarn.service.utils.ServiceRegistryUtils;
import org.apache.hadoop.yarn.service.utils.ServiceUtils;
import org.apache.hadoop.yarn.service.utils.SliderFileSystem;
import org.apache.hadoop.yarn.service.utils.ZookeeperUtils;
import org.apache.hadoop.yarn.util.DockerClientConfigHandler;
import org.apache.hadoop.yarn.util.Records;
import org.apache.hadoop.yarn.util.Times;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.text.MessageFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static org.apache.hadoop.yarn.api.records.YarnApplicationState.*;
import static org.apache.hadoop.yarn.service.conf.YarnServiceConf.*;
import static org.apache.hadoop.yarn.service.utils.ServiceApiUtil.jsonSerDeser;
import static org.apache.hadoop.yarn.service.utils.ServiceUtils.*;

@InterfaceAudience.Public
@InterfaceStability.Unstable
public class ServiceClient extends AppAdminClient implements SliderExitCodes,
    YarnServiceConstants {
  private static final Logger LOG =
      LoggerFactory.getLogger(ServiceClient.class);
  private SliderFileSystem fs;
  //TODO disable retry so that client / rest API doesn't block?
  protected YarnClient yarnClient;
  // Avoid looking up applicationId from fs all the time.
  private Map<String, AppInfo> cachedAppInfo = new ConcurrentHashMap<>();

  private RegistryOperations registryClient;
  private CuratorFramework curatorClient;
  private YarnRPC rpc;

  private static EnumSet<YarnApplicationState> terminatedStates =
      EnumSet.of(FINISHED, FAILED, KILLED);
  private static EnumSet<YarnApplicationState> liveStates =
      EnumSet.of(NEW, NEW_SAVING, SUBMITTED, ACCEPTED, RUNNING);
  private static EnumSet<YarnApplicationState> preRunningStates =
      EnumSet.of(NEW, NEW_SAVING, SUBMITTED, ACCEPTED);

  @Override protected void serviceInit(Configuration configuration)
      throws Exception {
    fs = new SliderFileSystem(configuration);
    yarnClient = YarnClient.createYarnClient();
    rpc = YarnRPC.create(configuration);
    addService(yarnClient);
    super.serviceInit(configuration);
  }

  @Override
  protected void serviceStop() throws Exception {
    if (registryClient != null) {
      registryClient.stop();
    }
    super.serviceStop();
  }

  public Service loadAppJsonFromLocalFS(String fileName, String serviceName,
      Long lifetime, String queue) throws IOException, YarnException {
    File file = new File(fileName);
    if (!file.exists() && fileName.equals(file.getName())) {
      String examplesDirStr = System.getenv("YARN_SERVICE_EXAMPLES_DIR");
      String[] examplesDirs;
      if (examplesDirStr == null) {
        String yarnHome = System
            .getenv(ApplicationConstants.Environment.HADOOP_YARN_HOME.key());
        examplesDirs = new String[]{
            yarnHome + "/share/hadoop/yarn/yarn-service-examples",
            yarnHome + "/yarn-service-examples"
        };
      } else {
        examplesDirs = StringUtils.split(examplesDirStr, ":");
      }
      for (String dir : examplesDirs) {
        file = new File(MessageFormat.format("{0}/{1}/{2}.json",
            dir, fileName, fileName));
        if (file.exists()) {
          break;
        }
        // Then look for secondary location.
        file = new File(MessageFormat.format("{0}/{1}.json",
            dir, fileName));
        if (file.exists()) {
          break;
        }
      }
    }
    if (!file.exists()) {
      throw new YarnException("File or example could not be found: " +
          fileName);
    }
    Path filePath = new Path(file.getAbsolutePath());
    LOG.info("Loading service definition from local FS: " + filePath);
    Service service = jsonSerDeser
        .load(FileSystem.getLocal(getConfig()), filePath);
    if (!StringUtils.isEmpty(serviceName)) {
      service.setName(serviceName);
    }
    if (lifetime != null && lifetime > 0) {
      service.setLifetime(lifetime);
    }
    if (!StringUtils.isEmpty(queue)) {
      service.setQueue(queue);
    }
    return service;
  }

  @Override
  public int actionSave(String fileName, String serviceName, Long lifetime,
      String queue) throws IOException, YarnException {
    return actionBuild(loadAppJsonFromLocalFS(fileName, serviceName,
        lifetime, queue));
  }

  public int actionBuild(Service service)
      throws YarnException, IOException {
    ServiceApiUtil.validateAndResolveService(service, fs, getConfig());
    Path appDir = checkAppNotExistOnHdfs(service, false);
    ServiceApiUtil.createDirAndPersistApp(fs, appDir, service);
    return EXIT_SUCCESS;
  }

  @Override
  public int actionUpgrade(String appName, String fileName)
      throws IOException, YarnException {
    checkAppExistOnHdfs(appName);
    Service upgradeService = loadAppJsonFromLocalFS(fileName, appName,
        null, null);
    return actionUpgrade(upgradeService);
  }

  public int actionUpgrade(Service service) throws YarnException, IOException {
    Service persistedService =
        ServiceApiUtil.loadService(fs, service.getName());
    if (!StringUtils.isEmpty(persistedService.getId())) {
      cachedAppInfo.put(persistedService.getName(), new AppInfo(
          ApplicationId.fromString(persistedService.getId()),
          persistedService.getKerberosPrincipal().getPrincipalName()));
    }

    if (persistedService.getVersion().equals(service.getVersion())) {
      String message =
          service.getName() + " is already at version " + service.getVersion()
              + ". There is nothing to upgrade.";
      LOG.error(message);
      throw new YarnException(message);
    }

    Path serviceUpgradeDir = checkAppNotExistOnHdfs(service, true);
    ServiceApiUtil.validateAndResolveService(service, fs, getConfig());
    ServiceApiUtil.createDirAndPersistApp(fs, serviceUpgradeDir, service);

    ApplicationReport appReport =
        yarnClient.getApplicationReport(getAppId(service.getName()));
    if (StringUtils.isEmpty(appReport.getHost())) {
      throw new YarnException(service.getName() + " AM hostname is empty");
    }
    ClientAMProtocol proxy = createAMProxy(service.getName(), appReport);

    UpgradeServiceRequestProto.Builder requestBuilder =
        UpgradeServiceRequestProto.newBuilder();
    requestBuilder.setVersion(service.getVersion());

    proxy.upgrade(requestBuilder.build());
    return EXIT_SUCCESS;
  }

  public int actionLaunch(String fileName, String serviceName, Long lifetime,
      String queue) throws IOException, YarnException {
    actionCreate(loadAppJsonFromLocalFS(fileName, serviceName, lifetime,
        queue));
    return EXIT_SUCCESS;
  }

  public ApplicationId actionCreate(Service service)
      throws IOException, YarnException {
    String serviceName = service.getName();
    ServiceApiUtil.validateAndResolveService(service, fs, getConfig());
    verifyNoLiveAppInRM(serviceName, "create");
    Path appDir = checkAppNotExistOnHdfs(service, false);

    // Write the definition first and then submit - AM will read the definition
    ServiceApiUtil.createDirAndPersistApp(fs, appDir, service);
    ApplicationId appId = submitApp(service);
    cachedAppInfo.put(serviceName, new AppInfo(appId, service
        .getKerberosPrincipal().getPrincipalName()));
    service.setId(appId.toString());
    // update app definition with appId
    ServiceApiUtil.writeAppDefinition(fs, appDir, service);
    return appId;
  }

  public int actionFlex(String serviceName, Map<String, String>
      componentCountStrings) throws YarnException, IOException {
    Map<String, Long> componentCounts =
        new HashMap<>(componentCountStrings.size());
    Service persistedService =
        ServiceApiUtil.loadService(fs, serviceName);
    if (!StringUtils.isEmpty(persistedService.getId())) {
      cachedAppInfo.put(persistedService.getName(), new AppInfo(
          ApplicationId.fromString(persistedService.getId()),
          persistedService.getKerberosPrincipal().getPrincipalName()));
    } else {
      throw new YarnException(persistedService.getName()
          + " appId is null, may be not submitted to YARN yet");
    }

    for (Map.Entry<String, String> entry : componentCountStrings.entrySet()) {
      String compName = entry.getKey();
      ServiceApiUtil.validateNameFormat(compName, getConfig());
      Component component = persistedService.getComponent(compName);
      if (component == null) {
        throw new IllegalArgumentException(entry.getKey() + " does not exist !");
      }
      long numberOfContainers =
          parseNumberOfContainers(component, entry.getValue());
      componentCounts.put(compName, numberOfContainers);
    }
    flexComponents(serviceName, componentCounts, persistedService);
    return EXIT_SUCCESS;
  }

  // Parse the number of containers requested by user, e.g.
  // +5 means add 5 additional containers
  // -5 means reduce 5 containers, if it goes to negative, sets it to 0
  // 5 means sets it to 5 containers.
  private long parseNumberOfContainers(Component component, String newNumber) {

    long orig = component.getNumberOfContainers();
    if (newNumber.startsWith("+")) {
      return orig + Long.parseLong(newNumber.substring(1));
    } else if (newNumber.startsWith("-")) {
      long ret = orig - Long.parseLong(newNumber.substring(1));
      if (ret < 0) {
        LOG.warn(MessageFormat.format(
            "[COMPONENT {0}]: component count goes to negative ({1}{2} = {3}),"
                + " ignore and reset it to 0.",
            component.getName(), orig, newNumber, ret));
        ret = 0;
      }
      return ret;
    } else {
      return Long.parseLong(newNumber);
    }
  }

  // Called by Rest Service
  public Map<String, Long> flexByRestService(String serviceName,
      Map<String, Long> componentCounts) throws YarnException, IOException {
    // load app definition
    Service persistedService = ServiceApiUtil.loadService(fs, serviceName);
    if (StringUtils.isEmpty(persistedService.getId())) {
      throw new YarnException(
          serviceName + " appId is null, may be not submitted to YARN yet");
    }
    cachedAppInfo.put(persistedService.getName(), new AppInfo(
        ApplicationId.fromString(persistedService.getId()), persistedService
        .getKerberosPrincipal().getPrincipalName()));
    return flexComponents(serviceName, componentCounts, persistedService);
  }

  private Map<String, Long> flexComponents(String serviceName,
      Map<String, Long> componentCounts, Service persistedService)
      throws YarnException, IOException {
    ServiceApiUtil.validateNameFormat(serviceName, getConfig());

    Map<String, Long> original = new HashMap<>(componentCounts.size());

    ComponentCountProto.Builder countBuilder = ComponentCountProto.newBuilder();
    FlexComponentsRequestProto.Builder requestBuilder =
        FlexComponentsRequestProto.newBuilder();

    for (Component persistedComp : persistedService.getComponents()) {
      String name = persistedComp.getName();
      if (componentCounts.containsKey(persistedComp.getName())) {
        original.put(name, persistedComp.getNumberOfContainers());
        persistedComp.setNumberOfContainers(componentCounts.get(name));

        // build the request
        countBuilder.setName(persistedComp.getName())
            .setNumberOfContainers(persistedComp.getNumberOfContainers());
        requestBuilder.addComponents(countBuilder.build());
      }
    }
    if (original.size() < componentCounts.size()) {
      componentCounts.keySet().removeAll(original.keySet());
      throw new YarnException("Components " + componentCounts.keySet()
          + " do not exist in app definition.");
    }
    jsonSerDeser
        .save(fs.getFileSystem(), ServiceApiUtil.getServiceJsonPath(fs, serviceName),
            persistedService, true);

    ApplicationId appId = getAppId(serviceName);
    if (appId == null) {
      String message = "Application ID doesn't exist for " + serviceName;
      LOG.error(message);
      throw new YarnException(message);
    }
    ApplicationReport appReport =
        yarnClient.getApplicationReport(appId);
    if (appReport.getYarnApplicationState() != RUNNING) {
      String message =
          serviceName + " is at " + appReport.getYarnApplicationState()
              + " state, flex can only be invoked when service is running";
      LOG.error(message);
      throw new YarnException(message);
    }
    if (StringUtils.isEmpty(appReport.getHost())) {
      throw new YarnException(serviceName + " AM hostname is empty");
    }
    ClientAMProtocol proxy =
        createAMProxy(serviceName, appReport);
    proxy.flexComponents(requestBuilder.build());
    for (Map.Entry<String, Long> entry : original.entrySet()) {
      LOG.info("[COMPONENT {}]: number of containers changed from {} to {}",
          entry.getKey(), entry.getValue(),
          componentCounts.get(entry.getKey()));
    }
    return original;
  }

  @Override
  public int actionStop(String serviceName)
      throws YarnException, IOException {
    return actionStop(serviceName, true);
  }

  public int actionStop(String serviceName, boolean waitForAppStopped)
      throws YarnException, IOException {
    ServiceApiUtil.validateNameFormat(serviceName, getConfig());
    ApplicationId currentAppId = getAppId(serviceName);
    if (currentAppId == null) {
      LOG.info("Application ID doesn't exist for service {}", serviceName);
      cleanUpRegistry(serviceName);
      return EXIT_COMMAND_ARGUMENT_ERROR;
    }
    ApplicationReport report = yarnClient.getApplicationReport(currentAppId);
    if (terminatedStates.contains(report.getYarnApplicationState())) {
      LOG.info("Service {} is already in a terminated state {}", serviceName,
          report.getYarnApplicationState());
      cleanUpRegistry(serviceName);
      return EXIT_COMMAND_ARGUMENT_ERROR;
    }
    if (preRunningStates.contains(report.getYarnApplicationState())) {
      String msg = serviceName + " is at " + report.getYarnApplicationState()
          + ", forcefully killed by user!";
      yarnClient.killApplication(currentAppId, msg);
      LOG.info(msg);
      cleanUpRegistry(serviceName);
      return EXIT_SUCCESS;
    }
    if (StringUtils.isEmpty(report.getHost())) {
      throw new YarnException(serviceName + " AM hostname is empty");
    }
    LOG.info("Stopping service {}, with appId = {}", serviceName, currentAppId);
    try {
      ClientAMProtocol proxy =
          createAMProxy(serviceName, report);
      cachedAppInfo.remove(serviceName);
      if (proxy != null) {
        // try to stop the app gracefully.
        StopRequestProto request = StopRequestProto.newBuilder().build();
        proxy.stop(request);
        LOG.info("Service " + serviceName + " is being gracefully stopped...");
      } else {
        yarnClient.killApplication(currentAppId,
            serviceName + " is forcefully killed by user!");
        LOG.info("Forcefully kill the service: " + serviceName);
        cleanUpRegistry(serviceName);
        return EXIT_SUCCESS;
      }

      if (!waitForAppStopped) {
        cleanUpRegistry(serviceName);
        return EXIT_SUCCESS;
      }
      // Wait until the app is killed.
      long startTime = System.currentTimeMillis();
      int pollCount = 0;
      while (true) {
        Thread.sleep(2000);
        report = yarnClient.getApplicationReport(currentAppId);
        if (terminatedStates.contains(report.getYarnApplicationState())) {
          LOG.info("Service " + serviceName + " is stopped.");
          break;
        }
        // Forcefully kill after 10 seconds.
        if ((System.currentTimeMillis() - startTime) > 10000) {
          LOG.info("Stop operation timeout stopping, forcefully kill the app "
              + serviceName);
          yarnClient.killApplication(currentAppId,
              "Forcefully kill the app by user");
          break;
        }
        if (++pollCount % 10 == 0) {
          LOG.info("Waiting for service " + serviceName + " to be stopped.");
        }
      }
    } catch (IOException | YarnException | InterruptedException e) {
      LOG.info("Failed to stop " + serviceName + " gracefully due to: "
          + e.getMessage() + ", forcefully kill the app.");
      yarnClient.killApplication(currentAppId, "Forcefully kill the app");
    }
    cleanUpRegistry(serviceName);
    return EXIT_SUCCESS;
  }

  @Override
  public int actionDestroy(String serviceName) throws YarnException,
      IOException {
    ServiceApiUtil.validateNameFormat(serviceName, getConfig());
    verifyNoLiveAppInRM(serviceName, "destroy");

    Path appDir = fs.buildClusterDirPath(serviceName);
    FileSystem fileSystem = fs.getFileSystem();
    // remove from the appId cache
    cachedAppInfo.remove(serviceName);
    int ret = EXIT_SUCCESS;
    if (fileSystem.exists(appDir)) {
      if (fileSystem.delete(appDir, true)) {
        LOG.info("Successfully deleted service dir for " + serviceName + ": "
            + appDir);
      } else {
        String message =
            "Failed to delete service + " + serviceName + " at:  " + appDir;
        LOG.info(message);
        throw new YarnException(message);
      }
    } else {
      LOG.info("Service '" + serviceName + "' doesn't exist at hdfs path: "
          + appDir);
      ret = EXIT_NOT_FOUND;
    }
    try {
      deleteZKNode(serviceName);
      // don't set destroySucceed to false if no ZK node exists because not
      // all services use a ZK node
    } catch (Exception e) {
      throw new IOException("Could not delete zk node for " + serviceName, e);
    }
    if (!cleanUpRegistry(serviceName)) {
      if (ret == EXIT_SUCCESS) {
        ret = EXIT_OTHER_FAILURE;
      }
    }
    if (ret == EXIT_SUCCESS) {
      LOG.info("Successfully destroyed service {}", serviceName);
      return ret;
    } else if (ret == EXIT_NOT_FOUND) {
      LOG.error("Error on destroy '" + serviceName + "': not found.");
      return ret;
    } else {
      LOG.error("Error on destroy '" + serviceName + "': error cleaning up " +
          "registry.");
      return ret;
    }
  }

  private boolean cleanUpRegistry(String serviceName) throws SliderException {
    String registryPath =
        ServiceRegistryUtils.registryPathForInstance(serviceName);
    try {
      if (getRegistryClient().exists(registryPath)) {
        getRegistryClient().delete(registryPath, true);
      } else {
        LOG.info(
            "Service '" + serviceName + "' doesn't exist at ZK registry path: "
                + registryPath);
        // not counted as a failure if the registry entries don't exist
      }
    } catch (IOException e) {
      LOG.warn("Error deleting registry entry {}", registryPath, e);
      return false;
    }
    return true;
  }

  private synchronized RegistryOperations getRegistryClient()
      throws SliderException, IOException {

    if (registryClient == null) {
      registryClient =
          RegistryOperationsFactory.createInstance("ServiceClient", getConfig());
      registryClient.init(getConfig());
      registryClient.start();
    }
    return registryClient;
  }

  /**
   * Delete service's ZK node. This is a different node from the service's
   * registry entry and is set aside for the service to use for its own ZK data.
   *
   * @param serviceName service name
   * @return true if the node was deleted, false if the node doesn't exist
   * @throws Exception if the node couldn't be deleted
   */
  private boolean deleteZKNode(String serviceName) throws Exception {
    CuratorFramework curatorFramework = getCuratorClient();
    String user = RegistryUtils.currentUser();
    String zkPath = ServiceRegistryUtils.mkServiceHomePath(user, serviceName);
    if (curatorFramework.checkExists().forPath(zkPath) != null) {
      curatorFramework.delete().deletingChildrenIfNeeded().forPath(zkPath);
      LOG.info("Deleted zookeeper path: " + zkPath);
      return true;
    } else {
      LOG.info(
          "Service '" + serviceName + "' doesn't exist at ZK path: " + zkPath);
      return false;
    }
  }

  private synchronized CuratorFramework getCuratorClient()
      throws BadConfigException {
    String registryQuorum =
        getConfig().get(RegistryConstants.KEY_REGISTRY_ZK_QUORUM);

    // though if neither is set: trouble
    if (ServiceUtils.isUnset(registryQuorum)) {
      throw new BadConfigException(
          "No Zookeeper quorum provided in the" + " configuration property "
              + RegistryConstants.KEY_REGISTRY_ZK_QUORUM);
    }
    ZookeeperUtils.splitToHostsAndPortsStrictly(registryQuorum);

    if (curatorClient == null) {
      curatorClient =
          CuratorFrameworkFactory.builder().connectString(registryQuorum)
              .sessionTimeoutMs(10000).retryPolicy(new RetryNTimes(5, 2000))
              .build();
      curatorClient.start();
    }
    return curatorClient;
  }

  private void verifyNoLiveAppInRM(String serviceName, String action)
      throws IOException, YarnException {
    Set<String> types = new HashSet<>(1);
    types.add(YarnServiceConstants.APP_TYPE);
    Set<String> tags = null;
    if (serviceName != null) {
      tags = Collections.singleton(ServiceUtils.createNameTag(serviceName));
    }
    GetApplicationsRequest request = GetApplicationsRequest.newInstance();
    request.setApplicationTypes(types);
    request.setApplicationTags(tags);
    request.setApplicationStates(liveStates);
    String user = UserGroupInformation.getCurrentUser().getUserName();
    if (user != null) {
      request.setUsers(Collections.singleton(user));
    }
    List<ApplicationReport> reports = yarnClient.getApplications(request);
    if (!reports.isEmpty()) {
      String message = "";
      if (action.equals("destroy")) {
        message = "Failed to destroy service " + serviceName
            + ", because it is still running.";
      } else {
        message = "Failed to " + action + " service " + serviceName
            + ", because it already exists.";
      }
      throw new YarnException(message);
    }
  }

  @VisibleForTesting
  ApplicationId submitApp(Service app) throws IOException, YarnException {
    String serviceName = app.getName();
    Configuration conf = getConfig();
    Path appRootDir = fs.buildClusterDirPath(app.getName());

    YarnClientApplication yarnApp = yarnClient.createApplication();
    ApplicationSubmissionContext submissionContext =
        yarnApp.getApplicationSubmissionContext();
    ServiceApiUtil.validateCompResourceSize(
        yarnApp.getNewApplicationResponse().getMaximumResourceCapability(),
        app);

    submissionContext.setKeepContainersAcrossApplicationAttempts(true);
    if (app.getLifetime() > 0) {
      Map<ApplicationTimeoutType, Long> appTimeout = new HashMap<>();
      appTimeout.put(ApplicationTimeoutType.LIFETIME, app.getLifetime());
      submissionContext.setApplicationTimeouts(appTimeout);
    }
    submissionContext.setMaxAppAttempts(YarnServiceConf
        .getInt(YarnServiceConf.AM_RESTART_MAX, DEFAULT_AM_RESTART_MAX, app
            .getConfiguration(), conf));

    setLogAggregationContext(app, conf, submissionContext);

    Map<String, LocalResource> localResources = new HashMap<>();

    // copy local slideram-log4j.properties to hdfs and add to localResources
    boolean hasAMLog4j =
        addAMLog4jResource(serviceName, conf, localResources);
    // copy jars to hdfs and add to localResources
    addJarResource(serviceName, localResources);
    // add keytab if in secure env
    addKeytabResourceIfSecure(fs, localResources, app);
    if (LOG.isDebugEnabled()) {
      printLocalResources(localResources);
    }
    Map<String, String> env = addAMEnv();

    // create AM CLI
    String cmdStr = buildCommandLine(app, conf, appRootDir, hasAMLog4j);
    submissionContext.setResource(Resource.newInstance(YarnServiceConf
        .getLong(YarnServiceConf.AM_RESOURCE_MEM,
            YarnServiceConf.DEFAULT_KEY_AM_RESOURCE_MEM, app.getConfiguration(),
            conf), 1));
    String queue = app.getQueue();
    if (StringUtils.isEmpty(queue)) {
      queue = conf.get(YARN_QUEUE, DEFAULT_YARN_QUEUE);
    }
    submissionContext.setQueue(queue);
    submissionContext.setApplicationName(serviceName);
    submissionContext.setApplicationType(YarnServiceConstants.APP_TYPE);
    Set<String> appTags =
        AbstractClientProvider.createApplicationTags(serviceName, null, null);
    if (!appTags.isEmpty()) {
      submissionContext.setApplicationTags(appTags);
    }
    ContainerLaunchContext amLaunchContext =
        Records.newRecord(ContainerLaunchContext.class);
    amLaunchContext.setCommands(Collections.singletonList(cmdStr));
    amLaunchContext.setEnvironment(env);
    amLaunchContext.setLocalResources(localResources);
    addCredentials(amLaunchContext, app);
    submissionContext.setAMContainerSpec(amLaunchContext);
    yarnClient.submitApplication(submissionContext);
    return submissionContext.getApplicationId();
  }

  private void setLogAggregationContext(Service app, Configuration conf,
      ApplicationSubmissionContext submissionContext) {
    LogAggregationContext context = Records.newRecord(LogAggregationContext
        .class);
    String finalLogInclude = YarnServiceConf.get
        (FINAL_LOG_INCLUSION_PATTERN, null, app.getConfiguration(), conf);
    if (!StringUtils.isEmpty(finalLogInclude)) {
      context.setIncludePattern(finalLogInclude);
    }
    String finalLogExclude = YarnServiceConf.get
        (FINAL_LOG_EXCLUSION_PATTERN, null, app.getConfiguration(), conf);
    if (!StringUtils.isEmpty(finalLogExclude)) {
      context.setExcludePattern(finalLogExclude);
    }
    String rollingLogInclude = YarnServiceConf.get
        (ROLLING_LOG_INCLUSION_PATTERN, null, app.getConfiguration(), conf);
    if (!StringUtils.isEmpty(rollingLogInclude)) {
      context.setRolledLogsIncludePattern(rollingLogInclude);
    }
    String rollingLogExclude = YarnServiceConf.get
        (ROLLING_LOG_EXCLUSION_PATTERN, null, app.getConfiguration(), conf);
    if (!StringUtils.isEmpty(rollingLogExclude)) {
      context.setRolledLogsExcludePattern(rollingLogExclude);
    }
    submissionContext.setLogAggregationContext(context);
  }

  private void printLocalResources(Map<String, LocalResource> map) {
    LOG.debug("Added LocalResource for localization: ");
    StringBuilder builder = new StringBuilder();
    for (Map.Entry<String, LocalResource> entry : map.entrySet()) {
      builder.append(entry.getKey()).append(" -> ")
          .append(entry.getValue().getResource().getFile())
          .append(System.lineSeparator());
    }
    LOG.debug(builder.toString());
  }

  private String buildCommandLine(Service app, Configuration conf,
      Path appRootDir, boolean hasSliderAMLog4j) throws BadConfigException {
    JavaCommandLineBuilder CLI = new JavaCommandLineBuilder();
    CLI.forceIPv4().headless();
    String jvmOpts = YarnServiceConf
        .get(YarnServiceConf.JVM_OPTS, "", app.getConfiguration(), conf);
    if (!jvmOpts.contains("-Xmx")) {
      jvmOpts += DEFAULT_AM_JVM_XMX;
    }

    CLI.setJVMOpts(jvmOpts);
    if (hasSliderAMLog4j) {
      CLI.sysprop(SYSPROP_LOG4J_CONFIGURATION, YARN_SERVICE_LOG4J_FILENAME);
      CLI.sysprop(SYSPROP_LOG_DIR, ApplicationConstants.LOG_DIR_EXPANSION_VAR);
    }
    CLI.add(ServiceMaster.class.getCanonicalName());
    //TODO debugAM CLI.add(Arguments.ARG_DEBUG)
    CLI.add("-" + ServiceMaster.YARNFILE_OPTION, new Path(appRootDir,
        app.getName() + ".json"));
    // pass the registry binding
    CLI.addConfOptionToCLI(conf, RegistryConstants.KEY_REGISTRY_ZK_ROOT,
        RegistryConstants.DEFAULT_ZK_REGISTRY_ROOT);
    CLI.addMandatoryConfOption(conf, RegistryConstants.KEY_REGISTRY_ZK_QUORUM);

    // write out the path output
    CLI.addOutAndErrFiles(STDOUT_AM, STDERR_AM);
    String cmdStr = CLI.build();
    LOG.debug("AM launch command: {}", cmdStr);
    return cmdStr;
  }

  private Map<String, String> addAMEnv() throws IOException {
    Map<String, String> env = new HashMap<>();
    ClasspathConstructor classpath =
        buildClasspath(YarnServiceConstants.SUBMITTED_CONF_DIR, "lib", fs, getConfig()
            .getBoolean(YarnConfiguration.IS_MINI_YARN_CLUSTER, false));
    env.put("CLASSPATH", classpath.buildClasspath());
    env.put("LANG", "en_US.UTF-8");
    env.put("LC_ALL", "en_US.UTF-8");
    env.put("LANGUAGE", "en_US.UTF-8");
    String jaas = System.getenv("HADOOP_JAAS_DEBUG");
    if (jaas != null) {
      env.put("HADOOP_JAAS_DEBUG", jaas);
    }
    if (!UserGroupInformation.isSecurityEnabled()) {
      String userName = UserGroupInformation.getCurrentUser().getUserName();
      LOG.debug("Run as user " + userName);
      // HADOOP_USER_NAME env is used by UserGroupInformation when log in
      // This env makes AM run as this user
      env.put("HADOOP_USER_NAME", userName);
    }
    LOG.debug("AM env: \n{}", stringifyMap(env));
    return env;
  }

  protected Path addJarResource(String serviceName,
      Map<String, LocalResource> localResources)
      throws IOException, SliderException {
    Path libPath = fs.buildClusterDirPath(serviceName);
    ProviderUtils
        .addProviderJar(localResources, ServiceMaster.class, SERVICE_CORE_JAR, fs,
            libPath, "lib", false);
    Path dependencyLibTarGzip = fs.getDependencyTarGzip();
    if (fs.isFile(dependencyLibTarGzip)) {
      LOG.info("Loading lib tar from " + dependencyLibTarGzip);
      fs.submitTarGzipAndUpdate(localResources);
    } else {
      if (dependencyLibTarGzip != null) {
        LOG.warn("Property {} has a value {}, but is not a valid file",
            YarnServiceConf.DEPENDENCY_TARBALL_PATH, dependencyLibTarGzip);
      }
      String[] libs = ServiceUtils.getLibDirs();
      LOG.info("Uploading all dependency jars to HDFS. For faster submission of"
          + " apps, set config property {} to the dependency tarball location."
          + " Dependency tarball can be uploaded to any HDFS path directly"
          + " or by using command: yarn app -{} [<Destination Folder>]",
          YarnServiceConf.DEPENDENCY_TARBALL_PATH,
          ApplicationCLI.ENABLE_FAST_LAUNCH);
      for (String libDirProp : libs) {
        ProviderUtils.addAllDependencyJars(localResources, fs, libPath, "lib",
            libDirProp);
      }
    }
    return libPath;
  }

  private boolean addAMLog4jResource(String serviceName, Configuration conf,
      Map<String, LocalResource> localResources)
      throws IOException, BadClusterStateException {
    boolean hasAMLog4j = false;
    String hadoopConfDir =
        System.getenv(ApplicationConstants.Environment.HADOOP_CONF_DIR.name());
    if (hadoopConfDir != null) {
      File localFile =
          new File(hadoopConfDir, YarnServiceConstants.YARN_SERVICE_LOG4J_FILENAME);
      if (localFile.exists()) {
        Path localFilePath = createLocalPath(localFile);
        Path appDirPath = fs.buildClusterDirPath(serviceName);
        Path remoteConfPath =
            new Path(appDirPath, YarnServiceConstants.SUBMITTED_CONF_DIR);
        Path remoteFilePath =
            new Path(remoteConfPath, YarnServiceConstants.YARN_SERVICE_LOG4J_FILENAME);
        copy(conf, localFilePath, remoteFilePath);
        LocalResource localResource =
            fs.createAmResource(remoteConfPath, LocalResourceType.FILE);
        localResources.put(localFilePath.getName(), localResource);
        hasAMLog4j = true;
      } else {
        LOG.warn("AM log4j property file doesn't exist: " + localFile);
      }
    }
    return hasAMLog4j;
  }

  @Override
  public int actionStart(String serviceName) throws YarnException, IOException {
    ServiceApiUtil.validateNameFormat(serviceName, getConfig());
    Service liveService = getStatus(serviceName);
    if (liveService == null ||
        !liveService.getState().equals(ServiceState.UPGRADING)) {
      Path appDir = checkAppExistOnHdfs(serviceName);
      Service service = ServiceApiUtil.loadService(fs, serviceName);
      ServiceApiUtil.validateAndResolveService(service, fs, getConfig());
      // see if it is actually running and bail out;
      verifyNoLiveAppInRM(serviceName, "start");
      ApplicationId appId = submitApp(service);
      service.setId(appId.toString());
      // write app definition on to hdfs
      Path appJson = ServiceApiUtil.writeAppDefinition(fs, appDir, service);
      LOG.info("Persisted service " + service.getName() + " at " + appJson);
      return 0;
    } else {
      LOG.info("Finalize service {} upgrade");
      ApplicationReport appReport =
          yarnClient.getApplicationReport(getAppId(serviceName));
      if (StringUtils.isEmpty(appReport.getHost())) {
        throw new YarnException(serviceName + " AM hostname is empty");
      }
      ClientAMProtocol proxy = createAMProxy(serviceName, appReport);

      RestartServiceRequestProto.Builder requestBuilder =
          RestartServiceRequestProto.newBuilder();
      proxy.restart(requestBuilder.build());
      return 0;
    }
  }

  /**
   * Verifies that the service definition does not exist on hdfs.
   *
   * @param service   service
   * @param isUpgrade true for upgrades; false otherwise
   * @return path to the service definition..
   * @throws IOException
   * @throws SliderException
   */
  private Path checkAppNotExistOnHdfs(Service service, boolean isUpgrade)
      throws IOException, SliderException {
    Path appDir = !isUpgrade ? fs.buildClusterDirPath(service.getName()) :
        fs.buildClusterUpgradeDirPath(service.getName(), service.getVersion());
    fs.verifyDirectoryNonexistent(
        new Path(appDir, service.getName() + ".json"));
    return appDir;
  }

  /**
   * Verifies that the service exists on hdfs.
   * @param serviceName service name
   * @return path to the service definition.
   * @throws IOException
   * @throws SliderException
   */
  private Path checkAppExistOnHdfs(String serviceName)
      throws IOException, SliderException {
    Path appDir = fs.buildClusterDirPath(serviceName);
    fs.verifyPathExists(new Path(appDir, serviceName + ".json"));
    return appDir;
  }

  private void addCredentials(ContainerLaunchContext amContext, Service app)
      throws IOException {
    Credentials allCreds = new Credentials();
    // HDFS DT
    if (UserGroupInformation.isSecurityEnabled()) {
      String tokenRenewer = YarnClientUtils.getRmPrincipal(getConfig());
      if (StringUtils.isEmpty(tokenRenewer)) {
        throw new IOException(
            "Can't get Master Kerberos principal for the RM to use as renewer");
      }
      final org.apache.hadoop.security.token.Token<?>[] tokens =
          fs.getFileSystem().addDelegationTokens(tokenRenewer, allCreds);
      if (LOG.isDebugEnabled()) {
        if (tokens != null && tokens.length != 0) {
          for (Token<?> token : tokens) {
            LOG.debug("Got DT: " + token);
          }
        }
      }
    }

    if (!StringUtils.isEmpty(app.getDockerClientConfig())) {
      allCreds.addAll(DockerClientConfigHandler.readCredentialsFromConfigFile(
          new Path(app.getDockerClientConfig()), getConfig(), app.getName()));
    }

    if (allCreds.numberOfTokens() > 0) {
      DataOutputBuffer dob = new DataOutputBuffer();
      allCreds.writeTokenStorageToStream(dob);
      ByteBuffer tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
      amContext.setTokens(tokens);
    }
  }

  private void addKeytabResourceIfSecure(SliderFileSystem fileSystem,
      Map<String, LocalResource> localResource, Service service)
      throws IOException, YarnException {
    if (!UserGroupInformation.isSecurityEnabled()) {
      return;
    }
    String principalName = service.getKerberosPrincipal().getPrincipalName();
    if (StringUtils.isEmpty(principalName)) {
      LOG.warn("No Kerberos principal name specified for " + service.getName());
      return;
    }
    if(StringUtils.isEmpty(service.getKerberosPrincipal().getKeytab())) {
      LOG.warn("No Kerberos keytab specified for " + service.getName());
      return;
    }

    URI keytabURI;
    try {
      keytabURI = new URI(service.getKerberosPrincipal().getKeytab());
    } catch (URISyntaxException e) {
      throw new YarnException(e);
    }

    switch (keytabURI.getScheme()) {
    case "hdfs":
      Path keytabOnhdfs = new Path(keytabURI);
      if (!fileSystem.getFileSystem().exists(keytabOnhdfs)) {
        LOG.warn(service.getName() + "'s keytab (principalName = " +
            principalName + ") doesn't exist at: " + keytabOnhdfs);
        return;
      }
      LocalResource keytabRes =
          fileSystem.createAmResource(keytabOnhdfs, LocalResourceType.FILE);
      localResource.put(String.format(YarnServiceConstants.KEYTAB_LOCATION,
          service.getName()), keytabRes);
      LOG.debug("Adding " + service.getName() + "'s keytab for " +
          "localization, uri = " + keytabOnhdfs);
      break;
    case "file":
      LOG.debug("Using a keytab from localhost: " + keytabURI);
      break;
    default:
      LOG.warn("Unsupported URI scheme " + keytabURI);
      break;
    }
  }

  public String updateLifetime(String serviceName, long lifetime)
      throws YarnException, IOException {
    ApplicationId currentAppId = getAppId(serviceName);
    if (currentAppId == null) {
      throw new YarnException("Application ID not found for " + serviceName);
    }
    ApplicationReport report = yarnClient.getApplicationReport(currentAppId);
    if (report == null) {
      throw new YarnException("Service not found for " + serviceName);
    }
    ApplicationId appId = report.getApplicationId();
    LOG.info("Updating lifetime of an service: serviceName = " + serviceName
        + ", appId = " + appId + ", lifetime = " + lifetime);
    Map<ApplicationTimeoutType, String> map = new HashMap<>();
    String newTimeout =
        Times.formatISO8601(System.currentTimeMillis() + lifetime * 1000);
    map.put(ApplicationTimeoutType.LIFETIME, newTimeout);
    UpdateApplicationTimeoutsRequest request =
        UpdateApplicationTimeoutsRequest.newInstance(appId, map);
    yarnClient.updateApplicationTimeouts(request);
    LOG.info(
        "Successfully updated lifetime for an service: serviceName = " + serviceName
            + ", appId = " + appId + ". New expiry time in ISO8601 format is "
            + newTimeout);
    return newTimeout;
  }

  public ServiceState convertState(YarnApplicationState state) {
    switch (state) {
    case NEW:
    case NEW_SAVING:
    case SUBMITTED:
    case ACCEPTED:
      return ServiceState.ACCEPTED;
    case RUNNING:
      return ServiceState.STARTED;
    case FINISHED:
    case KILLED:
      return ServiceState.STOPPED;
    case FAILED:
      return ServiceState.FAILED;
    default:
      return ServiceState.ACCEPTED;
    }
  }

  @Override
  public String getStatusString(String appIdOrName)
      throws IOException, YarnException {
    try {
      // try parsing appIdOrName, if it succeeds, it means it's appId
      ApplicationId appId = ApplicationId.fromString(appIdOrName);
      return getStatusByAppId(appId);
    } catch (IllegalArgumentException e) {
      // not appId format, it could be appName.
      Service status = getStatus(appIdOrName);
      return ServiceApiUtil.jsonSerDeser.toJson(status);
    }
  }

  private String getStatusByAppId(ApplicationId appId)
      throws IOException, YarnException {
    ApplicationReport appReport =
        yarnClient.getApplicationReport(appId);

    if (appReport.getYarnApplicationState() != RUNNING) {
      return "";
    }
    if (StringUtils.isEmpty(appReport.getHost())) {
      return "";
    }
    ClientAMProtocol amProxy = createAMProxy(appReport.getName(), appReport);
    GetStatusResponseProto response =
        amProxy.getStatus(GetStatusRequestProto.newBuilder().build());
    return response.getStatus();
  }

  public Service getStatus(String serviceName)
      throws IOException, YarnException {
    ServiceApiUtil.validateNameFormat(serviceName, getConfig());
    Service appSpec = new Service();
    appSpec.setName(serviceName);
    ApplicationId currentAppId = getAppId(serviceName);
    if (currentAppId == null) {
      LOG.info("Service {} does not have an application ID", serviceName);
      return appSpec;
    }
    ApplicationReport appReport = yarnClient.getApplicationReport(currentAppId);
    appSpec.setState(convertState(appReport.getYarnApplicationState()));
    ApplicationTimeout lifetime =
        appReport.getApplicationTimeouts().get(ApplicationTimeoutType.LIFETIME);
    if (lifetime != null) {
      appSpec.setLifetime(lifetime.getRemainingTime());
    }

    if (appReport.getYarnApplicationState() != RUNNING) {
      LOG.info("Service {} is at {} state", serviceName,
          appReport.getYarnApplicationState());
      return appSpec;
    }
    if (StringUtils.isEmpty(appReport.getHost())) {
      LOG.warn(serviceName + " AM hostname is empty");
      return appSpec;
    }
    ClientAMProtocol amProxy =
        createAMProxy(serviceName, appReport);
    GetStatusResponseProto response =
        amProxy.getStatus(GetStatusRequestProto.newBuilder().build());
    appSpec = jsonSerDeser.fromJson(response.getStatus());
    if (lifetime != null) {
      appSpec.setLifetime(lifetime.getRemainingTime());
    }
    return appSpec;
  }

  public YarnClient getYarnClient() {
    return this.yarnClient;
  }

  public int enableFastLaunch(String destinationFolder)
      throws IOException, YarnException {
    return actionDependency(destinationFolder, true);
  }

  public int actionDependency(String destinationFolder, boolean overwrite)
      throws IOException, YarnException {
    String currentUser = RegistryUtils.currentUser();
    LOG.info("Running command as user {}", currentUser);

    if (destinationFolder == null) {
      destinationFolder = String.format(YarnServiceConstants.DEPENDENCY_DIR,
          VersionInfo.getVersion());
    }
    Path dependencyLibTarGzip = new Path(destinationFolder,
        YarnServiceConstants.DEPENDENCY_TAR_GZ_FILE_NAME
            + YarnServiceConstants.DEPENDENCY_TAR_GZ_FILE_EXT);

    // Check if dependency has already been uploaded, in which case log
    // appropriately and exit success (unless overwrite has been requested)
    if (fs.isFile(dependencyLibTarGzip) && !overwrite) {
      System.out.println(String.format(
          "Dependency libs are already uploaded to %s.", dependencyLibTarGzip
              .toUri()));
      return EXIT_SUCCESS;
    }

    String[] libDirs = ServiceUtils.getLibDirs();
    if (libDirs.length > 0) {
      File tempLibTarGzipFile = File.createTempFile(
          YarnServiceConstants.DEPENDENCY_TAR_GZ_FILE_NAME + "_",
          YarnServiceConstants.DEPENDENCY_TAR_GZ_FILE_EXT);
      // copy all jars
      tarGzipFolder(libDirs, tempLibTarGzipFile, createJarFilter());

      LOG.info("Version Info: " + VersionInfo.getBuildVersion());
      fs.copyLocalFileToHdfs(tempLibTarGzipFile, dependencyLibTarGzip,
          new FsPermission(YarnServiceConstants.DEPENDENCY_DIR_PERMISSIONS));
      LOG.info("To let apps use this tarball, in yarn-site set config property "
          + "{} to {}", YarnServiceConf.DEPENDENCY_TARBALL_PATH,
          dependencyLibTarGzip);
      return EXIT_SUCCESS;
    } else {
      return EXIT_FALSE;
    }
  }

  protected ClientAMProtocol createAMProxy(String serviceName,
      ApplicationReport appReport) throws IOException, YarnException {

    if (UserGroupInformation.isSecurityEnabled()) {
      if (!cachedAppInfo.containsKey(serviceName)) {
        Service persistedService  = ServiceApiUtil.loadService(fs, serviceName);
        cachedAppInfo.put(serviceName, new AppInfo(appReport.getApplicationId(),
            persistedService.getKerberosPrincipal().getPrincipalName()));
      }
      String principalName = cachedAppInfo.get(serviceName).principalName;
      // Inject the principal into hadoop conf, because Hadoop
      // SaslRpcClient#getServerPrincipal requires a config for the
      // principal
      if (!StringUtils.isEmpty(principalName)) {
        getConfig().set(PRINCIPAL, principalName);
      } else {
        throw new YarnException("No principal specified in the persisted " +
            "service definition, fail to connect to AM.");
      }
    }
    InetSocketAddress address =
        NetUtils.createSocketAddrForHost(appReport.getHost(), appReport
            .getRpcPort());
    return ClientAMProxy.createProxy(getConfig(), ClientAMProtocol.class,
        UserGroupInformation.getCurrentUser(), rpc, address);
  }

  @VisibleForTesting
  void setFileSystem(SliderFileSystem fileSystem)
      throws IOException {
    this.fs = fileSystem;
  }

  @VisibleForTesting
  void setYarnClient(YarnClient yarnClient) {
    this.yarnClient = yarnClient;
  }

  public synchronized ApplicationId getAppId(String serviceName)
      throws IOException, YarnException {
    if (cachedAppInfo.containsKey(serviceName)) {
      return cachedAppInfo.get(serviceName).appId;
    }
    Service persistedService = ServiceApiUtil.loadService(fs, serviceName);
    if (persistedService == null) {
      throw new YarnException("Service " + serviceName
          + " doesn't exist on hdfs. Please check if the app exists in RM");
    }
    if (persistedService.getId() == null) {
      return null;
    }
    ApplicationId currentAppId = ApplicationId.fromString(persistedService
        .getId());
    cachedAppInfo.put(serviceName, new AppInfo(currentAppId, persistedService
        .getKerberosPrincipal().getPrincipalName()));
    return currentAppId;
  }

  private static class AppInfo {
    ApplicationId appId;
    String principalName;

    AppInfo(ApplicationId appId, String principalName) {
      this.appId = appId;
      this.principalName = principalName;
    }
  }
}