/*
 * 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.giraph.yarn;

import com.google.common.collect.ImmutableList;

import com.google.common.collect.Maps;
import org.apache.giraph.conf.GiraphConfiguration;
import org.apache.giraph.conf.GiraphConstants;
import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.protocolrecords
  .RegisterApplicationMasterResponse;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
import org.apache.hadoop.yarn.client.api.async.NMClientAsync;
import org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;

import org.apache.log4j.Logger;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * The YARN Application Master for Giraph is launched when the GiraphYarnClient
 * successfully requests an execution container from the Resource Manager. The
 * Application Master is provided by Giraph to manage all requests for resources
 * (worker nodes, memory, jar files, job configuration metadata, etc.) that
 * Giraph will need to perform the job. When Giraph runs in a non-YARN context,
 * the role of the Application Master is played by Hadoop when it launches our
 * GraphMappers (worker/master task nodes) to run the job.
 */
public class GiraphApplicationMaster {
  /** Logger */
  private static final Logger LOG =
    Logger.getLogger(GiraphApplicationMaster.class);
  /** Exit code for YARN containers that were manually killed/aborted */
  private static final int YARN_ABORT_EXIT_STATUS = -100;
  /** Exit code for successfully run YARN containers */
  private static final int YARN_SUCCESS_EXIT_STATUS = 0;
  /** millis to sleep between heartbeats during long loops */
  private static final int SLEEP_BETWEEN_HEARTBEATS_MSECS = 900;
  /** A reusable map of resources already in HDFS for each task to copy-to-local
   * env and use to launch each GiraphYarnTask. */
  private static Map<String, LocalResource> LOCAL_RESOURCES;
  /** Initialize the Configuration class with the resource file exported by
   * the YarnClient. We will need to export this resource to the tasks also.
   * Construct the HEARTBEAT to use to ping the RM about job progress/health.
   */
//TODO
  /** For status update for clients - yet to be implemented\\
  * Hostname of the container
  */
  private String appMasterHostname = "";
  /** Port on which the app master listens for status updates from clients*/
  private int appMasterRpcPort = 0;
  /** Tracking url to which app master publishes info for clients to monitor*/
  private String appMasterTrackingUrl = "";

  static {
    // pick up new conf XML file and populate it with stuff exported from client
    Configuration.addDefaultResource(GiraphConstants.GIRAPH_YARN_CONF_FILE);
  }

  /** GiraphApplicationMaster's application attempt id */
  private final ApplicationAttemptId appAttemptId;
  /** GiraphApplicationMaster container id. Leave me here, I'm very useful */
  private final ContainerId containerId;
  /** number of containers Giraph needs (conf.getMaxWorkers() + 1 master) */
  private final int containersToLaunch;
  /** MB of JVM heap per Giraph task container */
  private final int heapPerContainer;
  /** Giraph configuration for this job, transported here by YARN framework */
  private final ImmutableClassesGiraphConfiguration giraphConf;
  /** Yarn configuration for this job*/
  private final YarnConfiguration yarnConf;
  /** Completed Containers Counter */
  private final AtomicInteger completedCount;
  /** Failed Containers Counter */
  private final AtomicInteger failedCount;
  /** Number of containers requested (hopefully '-w' from our conf) */
  private final AtomicInteger allocatedCount;
  /** Number of successfully completed containers in this job run. */
  private final AtomicInteger successfulCount;
  /** the ACK #'s for AllocateRequests + heartbeats == last response # */
  private AtomicInteger lastResponseId;
  /** buffer tostore all tokens */
  private ByteBuffer allTokens;
  /** Executor to attempt asynchronous launches of Giraph containers */
  private ExecutorService executor;
  /** YARN progress is a <code>float</code> between 0.0f and 1.0f */
  //Handle to communicate with the Resource Manager
  @SuppressWarnings("rawtypes")
  private AMRMClientAsync amRMClient;
  /** Handle to communicate with the Node Manager */
  private NMClientAsync nmClientAsync;
  /** Listen to process the response from the Node Manager */
  private NMCallbackHandler containerListener;
  /** whether all containers finishe */
  private volatile boolean done;

  /**
   * Construct the GiraphAppMaster, populate fields using env vars
   * set up by YARN framework in this execution container.
   * @param cId the ContainerId
   * @param aId the ApplicationAttemptId
   */
  protected GiraphApplicationMaster(ContainerId cId, ApplicationAttemptId aId)
    throws IOException {
    containerId = cId; // future good stuff will need me to operate.
    appAttemptId = aId;
    lastResponseId = new AtomicInteger(0);
    giraphConf =
      new ImmutableClassesGiraphConfiguration(new GiraphConfiguration());
    yarnConf = new YarnConfiguration(giraphConf);
    completedCount = new AtomicInteger(0);
    failedCount = new AtomicInteger(0);
    allocatedCount = new AtomicInteger(0);
    successfulCount = new AtomicInteger(0);
    containersToLaunch = giraphConf.getMaxWorkers() + 1;
    executor = Executors.newFixedThreadPool(containersToLaunch);
    heapPerContainer = giraphConf.getYarnTaskHeapMb();
    LOG.info("GiraphAM  for ContainerId " + cId + " ApplicationAttemptId " +
      aId);
  }

  /**
   * Coordinates all requests for Giraph's worker/master task containers, and
   * manages application liveness heartbeat, completion status, teardown, etc.
   * @return success or failure
   */
  private boolean run() throws YarnException, IOException {
    boolean success = false;
    try {
      getAllTokens();
      registerRMCallBackHandler();
      registerNMCallbackHandler();
      registerAMToRM();
      madeAllContainerRequestToRM();
      LOG.info("Wait to finish ..");
      while (!done) {
        try {
          Thread.sleep(200);
        } catch (InterruptedException ex) {
          LOG.error(ex);
          //TODO:
        }
      }
      LOG.info("Done " + done);
    } finally {
      // if we get here w/o problems, the executor is already long finished.
      if (null != executor && !executor.isTerminated()) {
        LOG.info("Forcefully terminating executors with done =:" + done);
        executor.shutdownNow(); // force kill, especially if got here by throw
      }
      success = finish();
    }
    return success;
  }

  /**
   * Call when the application is done
   * @return if all containers succeed
   */
  private boolean finish() {
    // When the application completes, it should stop all running containers
    LOG.info("Application completed. Stopping running containers");
    nmClientAsync.stop();

    // When the application completes, it should send a finish application
    // signal to the RM
    LOG.info("Application completed. Signalling finish to RM");
    FinalApplicationStatus appStatus;
    String appMessage = null;
    boolean success = true;
    if (failedCount.get() == 0 &&
        completedCount.get() == containersToLaunch) {
      appStatus = FinalApplicationStatus.SUCCEEDED;
    } else {
      appStatus = FinalApplicationStatus.FAILED;
      appMessage = "Diagnostics." + ", total=" + containersToLaunch +
        ", completed=" + completedCount.get() +  ", failed=" +
        failedCount.get();
      success = false;
    }
    try {
      amRMClient.unregisterApplicationMaster(appStatus, appMessage, null);
    } catch (YarnException ex) {
      LOG.error("Failed to unregister application", ex);
    } catch (IOException e) {
      LOG.error("Failed to unregister application", e);
    }

    amRMClient.stop();
    return success;
  }
  /**
   * Add all containers' request
   * @return
   */
  private void madeAllContainerRequestToRM() {
    // Setup ask for containers from RM
    // Send request for containers to RM
    // Until we get our fully allocated quota, we keep on polling RM for
    // containers
    // Keep looping until all the containers are launched and shell script
    // executed on them ( regardless of success/failure).
    for (int i = 0; i < containersToLaunch; ++i) {
      ContainerRequest containerAsk = setupContainerAskForRM();
      amRMClient.addContainerRequest(containerAsk);
    }
  }

   /**
    * Setup the request that will be sent to the RM for the container ask.
    *
    * @return the setup ResourceRequest to be sent to RM
    */
  private ContainerRequest setupContainerAskForRM() {
    // setup requirements for hosts
    // using * as any host will do for the distributed shell app
    // set the priority for the request
    Priority pri = Records.newRecord(Priority.class);
    // TODO - what is the range for priority? how to decide?
    pri.setPriority(GiraphConstants.GIRAPH_YARN_PRIORITY);

    // Set up resource type requirements
    // For now, only memory is supported so we set memory requirements
    Resource capability = Records.newRecord(Resource.class);
    capability.setMemory(heapPerContainer);

    ContainerRequest request = new ContainerRequest(capability, null, null,
      pri);
    LOG.info("Requested container ask: " + request.toString());
    return request;
  }

  /**
   * Populate allTokens with the tokens received
   * @return
   */
  private void getAllTokens() throws IOException {
    Credentials credentials = UserGroupInformation.getCurrentUser()
        .getCredentials();
    DataOutputBuffer dob = new DataOutputBuffer();
    credentials.writeTokenStorageToStream(dob);
    // Now remove the AM->RM token so that containers cannot access it.
    Iterator<Token<?>> iter = credentials.getAllTokens().iterator();
    while (iter.hasNext()) {
      Token<?> token = iter.next();
      if (LOG.isDebugEnabled()) {
        LOG.debug("Token type :" + token.getKind());
      }
      if (token.getKind().equals(AMRMTokenIdentifier.KIND_NAME)) {
        iter.remove();
      }
    }
    allTokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
  }

  /**
   * Register RM callback and start listening
   * @return
   */
  private void registerRMCallBackHandler() {
    AMRMClientAsync.CallbackHandler allocListener = new RMCallbackHandler();
    amRMClient = AMRMClientAsync.createAMRMClientAsync(1000,
      allocListener);
    amRMClient.init(yarnConf);
    amRMClient.start();
  }

  /**
   * Register NM callback and start listening
   * @return
   */
  private void registerNMCallbackHandler() {
    containerListener = new NMCallbackHandler();
    nmClientAsync = new NMClientAsyncImpl(containerListener);
    nmClientAsync.init(yarnConf);
    nmClientAsync.start();
  }
  /**
   * Register AM to RM
   * @return AM register response
   */
  private RegisterApplicationMasterResponse registerAMToRM()
    throws YarnException {
    // register Application Master with the YARN Resource Manager so we can
    // begin requesting resources.
    try {
      if (UserGroupInformation.isSecurityEnabled()) {
        LOG.info("SECURITY ENABLED ");
      }
      // TODO: provide actual call back details
      RegisterApplicationMasterResponse response = amRMClient
        .registerApplicationMaster(appMasterHostname
        , appMasterRpcPort, appMasterTrackingUrl);
      return response;
    } catch (IOException ioe) {
      throw new IllegalStateException(
        "GiraphApplicationMaster failed to register with RM.", ioe);
    }
  }

  /**
   * For each container successfully allocated, attempt to set up and launch
   * a Giraph worker/master task.
   * @param allocatedContainers the containers we have currently allocated.
   */
  private void startContainerLaunchingThreads(final List<Container>
    allocatedContainers) {
    for (Container allocatedContainer : allocatedContainers) {
      LOG.info("Launching command on a new container." +
        ", containerId=" + allocatedContainer.getId() +
        ", containerNode=" + allocatedContainer.getNodeId().getHost() +
        ":" + allocatedContainer.getNodeId().getPort() +
        ", containerNodeURI=" + allocatedContainer.getNodeHttpAddress() +
        ", containerResourceMemory=" +
        allocatedContainer.getResource().getMemory());
      // Launch and start the container on a separate thread to keep the main
      // thread unblocked as all containers may not be allocated at one go.
      LaunchContainerRunnable runnableLaunchContainer =
        new LaunchContainerRunnable(allocatedContainer, containerListener);
      executor.execute(runnableLaunchContainer);
    }
  }

  /**
   * Lazily compose the map of jar and file names to LocalResource records for
   * inclusion in GiraphYarnTask container requests. Can re-use the same map
   * as Giraph tasks need identical HDFS-based resources (jars etc.) to run.
   * @return the resource map for a ContainerLaunchContext
   */
  private synchronized Map<String, LocalResource> getTaskResourceMap() {
    // Set the local resources: just send the copies already in HDFS
    if (null == LOCAL_RESOURCES) {
      LOCAL_RESOURCES = Maps.newHashMap();
      try {
        // if you have to update the giraphConf for export to tasks, do it now
        updateGiraphConfForExport();
        YarnUtils.addFsResourcesToMap(LOCAL_RESOURCES, giraphConf,
          appAttemptId.getApplicationId());
      } catch (IOException ioe) {
        // fail fast, this container will never launch.
        throw new IllegalStateException("Could not configure the container" +
          "launch context for GiraphYarnTasks.", ioe);
      }
    }
    // else, return the prepopulated copy to reuse for each GiraphYarkTask
    return LOCAL_RESOURCES;
  }

  /**
   * If you're going to make ANY CHANGES to your local GiraphConfiguration
   * while running the GiraphApplicationMaster, put them here.
   * This method replaces the current XML file GiraphConfiguration
   * stored in HDFS with the copy you have modified locally in-memory.
   */
  private void updateGiraphConfForExport()
    throws IOException {
    // Giraph expects this MapReduce stuff
    giraphConf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID,
      appAttemptId.getAttemptId());
    // now republish the giraph-conf.xml in HDFS
    YarnUtils.exportGiraphConfiguration(giraphConf,
      appAttemptId.getApplicationId());
  }

  /**
   * Application entry point
   * @param args command-line args (set by GiraphYarnClient, if any)
   */
  public static void main(final String[] args) {
    boolean result = false;
    LOG.info("Starting GiraphAM ");
    String containerIdString =  System.getenv().get(
      Environment.CONTAINER_ID.name());
    if (containerIdString == null) {
      // container id should always be set in the env by the framework
      throw new IllegalArgumentException("ContainerId not found in env vars.");
    }
    ContainerId containerId = ConverterUtils.toContainerId(containerIdString);
    ApplicationAttemptId appAttemptId = containerId.getApplicationAttemptId();
    try {
      GiraphApplicationMaster giraphAppMaster =
        new GiraphApplicationMaster(containerId, appAttemptId);
      result = giraphAppMaster.run();
      // CHECKSTYLE: stop IllegalCatch
    } catch (Throwable t) {
      // CHECKSTYLE: resume IllegalCatch
      LOG.error("GiraphApplicationMaster caught a " +
                  "top-level exception in main.", t);
      System.exit(1);
    }
    if (result) {
      LOG.info("Giraph Application Master completed successfully. exiting");
      System.exit(0);
    } else {
      LOG.info("Giraph Application Master failed. exiting");
      System.exit(2);
    }
  }

  /**
   * Thread to connect to the {@link ContainerManager} and launch the container
   * that will house one of our Giraph worker (or master) tasks.
   */
  private class LaunchContainerRunnable implements Runnable {
    /** Allocated container */
    private Container container;
    /** NM listener */
    private NMCallbackHandler containerListener;

    /**
     * Constructor.
     * @param newGiraphTaskContainer Allocated container
     * @param containerListener container listener.
     */
    public LaunchContainerRunnable(final Container newGiraphTaskContainer,
      NMCallbackHandler containerListener) {
      this.container = newGiraphTaskContainer;
      this.containerListener = containerListener;
    }

    /**
     * Connects to CM, sets up container launch context
     * for shell command and eventually dispatches the container
     * start request to the CM.
     */
    public void run() {
      // Connect to ContainerManager
      // configure the launcher for the Giraph task it will host
      ContainerLaunchContext ctx = buildContainerLaunchContext();
      // request CM to start this container as spec'd in ContainerLaunchContext
      containerListener.addContainer(container.getId(), container);
      nmClientAsync.startContainerAsync(container, ctx);
    }

    /**
     * Boilerplate to set up the ContainerLaunchContext to tell the Container
     * Manager how to launch our Giraph task in the execution container we have
     * already allocated.
     * @return a populated ContainerLaunchContext object.
     */
    private ContainerLaunchContext buildContainerLaunchContext() {
      LOG.info("Setting up container launch container for containerid=" +
        container.getId());
      ContainerLaunchContext launchContext = Records
        .newRecord(ContainerLaunchContext.class);
      // args inject the CLASSPATH, heap MB, and TaskAttemptID for launched task
      final List<String> commands = generateShellExecCommand();
      LOG.info("Conatain launch Commands :" + commands.get(0));
      launchContext.setCommands(commands);
      // Set up tokens for the container too. We are
      // populating them mainly for NodeManagers to be able to download any
      // files in the distributed file-system. The tokens are otherwise also
      // useful in cases, for e.g., when one is running a
      // "hadoop dfs" like command
      launchContext.setTokens(allTokens.slice());

      // add user information to the job
      String jobUserName = "ERROR_UNKNOWN_USER";
      UserGroupInformation ugi = null;
      try {
        ugi = UserGroupInformation.getCurrentUser();
        jobUserName = ugi.getUserName();
      } catch (IOException ioe) {
        jobUserName =
          System.getenv(ApplicationConstants.Environment.USER.name());
      }
      //launchContext.setUser(jobUserName);
      LOG.info("Setting username in ContainerLaunchContext to: " + jobUserName);
      // Set the environment variables to inject into remote task's container
      buildEnvironment(launchContext);
      // Set the local resources: just send the copies already in HDFS
      launchContext.setLocalResources(getTaskResourceMap());
      return launchContext;
    }

    /**
     * Generates our command line string used to launch our Giraph tasks.
     * @return the BASH shell commands to launch the job.
     */
    private List<String> generateShellExecCommand() {
      return ImmutableList.of("java " +
        "-Xmx" + heapPerContainer + "M " +
        "-Xms" + heapPerContainer + "M " +
        "-cp .:${CLASSPATH} " +
        "org.apache.giraph.yarn.GiraphYarnTask " +
        appAttemptId.getApplicationId().getClusterTimestamp() + " " +
        appAttemptId.getApplicationId().getId() + " " +
        container.getId().getId() + " " +
        appAttemptId.getAttemptId() + " " +
        "1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR +
        "/task-" + container.getId().getId() + "-stdout.log " +
        "2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR +
        "/task-" + container.getId().getId() + "-stderr.log "
      );
    }

    /**
     * Utility to populate the environment vars we wish to inject into the new
     * containter's env when the Giraph BSP task is executed.
     * @param launchContext the launch context which will set our environment
     *                      vars in the app master's execution container.
     */
    private void buildEnvironment(final ContainerLaunchContext launchContext) {
      Map<String, String> classPathForEnv = Maps.<String, String>newHashMap();
      // pick up the local classpath so when we instantiate a Configuration
      // remotely, we also get the "mapred-site.xml" and "yarn-site.xml"
      YarnUtils.addLocalClasspathToEnv(classPathForEnv, giraphConf);
      // set this map of env vars into the launch context.
      launchContext.setEnvironment(classPathForEnv);
    }
  }

  /**
   * CallbackHandler to process RM async calls
   */
  private class RMCallbackHandler implements AMRMClientAsync.CallbackHandler {
    @SuppressWarnings("unchecked")
    @Override
    public void onContainersCompleted(List<ContainerStatus>
      completedContainers) {
      LOG.info("Got response from RM for container ask, completedCnt=" +
        completedContainers.size());
      for (ContainerStatus containerStatus : completedContainers) {
        LOG.info("Got container status for containerID=" +
          containerStatus.getContainerId() + ", state=" +
          containerStatus.getState() + ", exitStatus=" +
          containerStatus.getExitStatus() + ", diagnostics=" +
          containerStatus.getDiagnostics());
        switch (containerStatus.getExitStatus()) {
        case YARN_SUCCESS_EXIT_STATUS:
          successfulCount.incrementAndGet();
          break;
        case YARN_ABORT_EXIT_STATUS:
          break; // not success or fail
        default:
          failedCount.incrementAndGet();
          break;
        }
        completedCount.incrementAndGet();
      }

      if (completedCount.get() == containersToLaunch) {
        done = true;
        LOG.info("All container compeleted. done = " + done);
      } else {
        LOG.info("After completion of one conatiner. current status is:" +
          " completedCount :" + completedCount.get() +
          " containersToLaunch :" + containersToLaunch +
          " successfulCount :" + successfulCount.get() +
          " failedCount :" + failedCount.get());
      }
    }
    @Override
    public void onContainersAllocated(List<Container> allocatedContainers) {
      LOG.info("Got response from RM for container ask, allocatedCnt=" +
          allocatedContainers.size());
      allocatedCount.addAndGet(allocatedContainers.size());
      LOG.info("Total allocated # of container so far : " +
        allocatedCount.get() +
        " allocated out of " + containersToLaunch + " required.");
      startContainerLaunchingThreads(allocatedContainers);
    }

    @Override
    public void onShutdownRequest() {
      done = true;
    }

    @Override
    public void onNodesUpdated(List<NodeReport> updatedNodes) {
    }

    @Override
    public float getProgress() {
      // set progress to deliver to RM on next heartbeat
      float progress = (float) completedCount.get() /
          containersToLaunch;
      return progress;
    }

    @Override
    public void onError(Throwable e) {
      done = true;
      amRMClient.stop();
    }
  }

  /**
   * CallbackHandler to process NM async calls
   */
  private class NMCallbackHandler implements NMClientAsync.CallbackHandler {
    /** List of containers */
    private ConcurrentMap<ContainerId, Container> containers =
          new ConcurrentHashMap<ContainerId, Container>();

    /**
     * Add a container
     * @param containerId id of container
     * @param container container object
     * @return
     */
    public void addContainer(ContainerId containerId, Container container) {
      containers.putIfAbsent(containerId, container);
    }

    @Override
    public void onContainerStopped(ContainerId containerId) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Succeeded to stop Container " + containerId);
      }
      containers.remove(containerId);
    }

    @Override
    public void onContainerStatusReceived(ContainerId containerId,
        ContainerStatus containerStatus) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Container Status: id=" + containerId + ", status=" +
            containerStatus);
      }
    }

    @Override
    public void onContainerStarted(ContainerId containerId,
        Map<String, ByteBuffer> allServiceResponse) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Succeeded to start Container " + containerId);
      }
      Container container = containers.get(containerId);
      if (container != null) {
        nmClientAsync.getContainerStatusAsync(containerId,
          container.getNodeId());
      }
    }

    @Override
    public void onStartContainerError(ContainerId containerId, Throwable t) {
      LOG.error("Failed to start Container " + containerId, t);
      containers.remove(containerId);
    }

    @Override
    public void onGetContainerStatusError(
      ContainerId containerId, Throwable t) {
      LOG.error("Failed to query the status of Container " + containerId, t);
    }

    @Override
    public void onStopContainerError(ContainerId containerId, Throwable t) {
      LOG.error("Failed to stop Container " + containerId);
      containers.remove(containerId);
    }
  }
}