/*
 * 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.twill.internal.appmaster;

import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.base.Strings;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.DiscreteDomains;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multiset;
import com.google.common.collect.Ranges;
import com.google.common.collect.Sets;
import com.google.common.io.Files;
import com.google.common.io.InputSupplier;
import com.google.common.reflect.TypeToken;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.hadoop.io.Text;
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.records.ContainerId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.util.Records;
import org.apache.twill.api.Command;
import org.apache.twill.api.EventHandler;
import org.apache.twill.api.EventHandlerSpecification;
import org.apache.twill.api.LocalFile;
import org.apache.twill.api.ResourceReport;
import org.apache.twill.api.ResourceSpecification;
import org.apache.twill.api.RunId;
import org.apache.twill.api.RuntimeSpecification;
import org.apache.twill.api.TwillRunResources;
import org.apache.twill.api.TwillSpecification;
import org.apache.twill.common.Threads;
import org.apache.twill.filesystem.Location;
import org.apache.twill.internal.Configs;
import org.apache.twill.internal.Constants;
import org.apache.twill.internal.ContainerInfo;
import org.apache.twill.internal.DefaultTwillRunResources;
import org.apache.twill.internal.EnvKeys;
import org.apache.twill.internal.JvmOptions;
import org.apache.twill.internal.ProcessLauncher;
import org.apache.twill.internal.TwillContainerLauncher;
import org.apache.twill.internal.TwillRuntimeSpecification;
import org.apache.twill.internal.json.JvmOptionsCodec;
import org.apache.twill.internal.json.LocalFileCodec;
import org.apache.twill.internal.json.TwillRuntimeSpecificationAdapter;
import org.apache.twill.internal.state.Message;
import org.apache.twill.internal.utils.Instances;
import org.apache.twill.internal.yarn.AbstractYarnTwillService;
import org.apache.twill.internal.yarn.YarnAMClient;
import org.apache.twill.internal.yarn.YarnContainerInfo;
import org.apache.twill.internal.yarn.YarnContainerStatus;
import org.apache.twill.internal.yarn.YarnUtils;
import org.apache.twill.zookeeper.ZKClient;
import org.apache.twill.zookeeper.ZKClients;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * The class that acts as {@code ApplicationMaster} for Twill applications.
 */
public final class ApplicationMasterService extends AbstractYarnTwillService implements Supplier<ResourceReport> {

  private static final Logger LOG = LoggerFactory.getLogger(ApplicationMasterService.class);
  private static final Gson GSON = new Gson();

  // Copied from org.apache.hadoop.yarn.security.AMRMTokenIdentifier.KIND_NAME since it's missing in Hadoop-2.0
  private static final Text AMRM_TOKEN_KIND_NAME = new Text("YARN_AM_RM_TOKEN");

  private final RunId runId;
  private final ZKClient zkClient;
  private final TwillSpecification twillSpec;
  private final ApplicationMasterLiveNodeData amLiveNode;
  private final RunningContainers runningContainers;
  private final ExpectedContainers expectedContainers;
  private final YarnAMClient amClient;
  private final JvmOptions jvmOpts;
  private final int reservedMemory;
  private final EventHandler eventHandler;
  private final Location applicationLocation;
  private final PlacementPolicyManager placementPolicyManager;
  private final Map<String, Map<String, String>> environments;
  private final TwillRuntimeSpecification twillRuntimeSpec;

  private volatile boolean stopped;
  private Queue<RunnableContainerRequest> runnableContainerRequests;
  private ExecutorService instanceChangeExecutor;

  public ApplicationMasterService(RunId runId, ZKClient zkClient, TwillRuntimeSpecification twillRuntimeSpec,
                                  YarnAMClient amClient, Location applicationLocation) throws Exception {
    super(zkClient, runId, applicationLocation);

    this.runId = runId;
    this.twillRuntimeSpec = twillRuntimeSpec;
    this.zkClient = zkClient;
    this.applicationLocation = applicationLocation;
    this.amClient = amClient;
    this.credentials = createCredentials();
    this.jvmOpts = loadJvmOptions();
    this.reservedMemory = twillRuntimeSpec.getReservedMemory();
    this.twillSpec = twillRuntimeSpec.getTwillSpecification();
    this.placementPolicyManager = new PlacementPolicyManager(twillSpec.getPlacementPolicies());
    this.environments = getEnvironments();

    this.amLiveNode = new ApplicationMasterLiveNodeData(Integer.parseInt(System.getenv(EnvKeys.YARN_APP_ID)),
                                                        Long.parseLong(System.getenv(EnvKeys.YARN_APP_ID_CLUSTER_TIME)),
                                                        amClient.getContainerId().toString());

    expectedContainers = initExpectedContainers(twillSpec);
    runningContainers = initRunningContainers(amClient.getContainerId(), amClient.getHost());
    eventHandler = createEventHandler(twillSpec);
  }

  private JvmOptions loadJvmOptions() throws IOException {
    final File jvmOptsFile = new File(Constants.Files.JVM_OPTIONS);
    if (!jvmOptsFile.exists()) {
      return new JvmOptions(null, JvmOptions.DebugOptions.NO_DEBUG);
    }
    return JvmOptionsCodec.decode(new InputSupplier<Reader>() {
      @Override
      public Reader getInput() throws IOException {
        return new FileReader(jvmOptsFile);
      }
    });
  }

  @SuppressWarnings("unchecked")
  @Nullable
  private EventHandler createEventHandler(TwillSpecification twillSpec) {
    try {
      // Should be able to load by this class ClassLoader, as they packaged in the same jar.
      EventHandlerSpecification handlerSpec = twillSpec.getEventHandler();
      if (handlerSpec == null) {
        return null;
      }

      Class<?> handlerClass = getClass().getClassLoader().loadClass(handlerSpec.getClassName());
      Preconditions.checkArgument(EventHandler.class.isAssignableFrom(handlerClass),
                                  "Class {} does not implements {}",
                                  handlerClass, EventHandler.class.getName());
      return Instances.newInstance((Class<? extends EventHandler>) handlerClass);
    } catch (Exception e) {
      throw Throwables.propagate(e);
    }
  }

  private RunningContainers initRunningContainers(ContainerId appMasterContainerId,
                                                  String appMasterHost) throws Exception {
    TwillRunResources appMasterResources = new DefaultTwillRunResources(
      0,
      appMasterContainerId.toString(),
      Integer.parseInt(System.getenv(EnvKeys.YARN_CONTAINER_VIRTUAL_CORES)),
      Integer.parseInt(System.getenv(EnvKeys.YARN_CONTAINER_MEMORY_MB)),
      appMasterHost, null, null);
    String appId = appMasterContainerId.getApplicationAttemptId().getApplicationId().toString();
    return new RunningContainers(appId, appMasterResources, zkClient);
  }

  private ExpectedContainers initExpectedContainers(TwillSpecification twillSpec) {
    Map<String, Integer> expectedCounts = Maps.newHashMap();
    for (RuntimeSpecification runtimeSpec : twillSpec.getRunnables().values()) {
      expectedCounts.put(runtimeSpec.getName(), runtimeSpec.getResourceSpecification().getInstances());
    }
    return new ExpectedContainers(expectedCounts);
  }

  @Override
  public ResourceReport get() {
    return runningContainers.getResourceReport();
  }

  @Override
  protected void doStart() throws Exception {
    LOG.info("Start application master with spec: " +
               TwillRuntimeSpecificationAdapter.create().toJson(twillRuntimeSpec));

    // initialize the event handler, if it fails, it will fail the application.
    if (eventHandler != null) {
      eventHandler.initialize(new BasicEventHandlerContext(twillSpec.getEventHandler()));
    }

    instanceChangeExecutor = Executors.newSingleThreadExecutor(Threads.createDaemonThreadFactory("instanceChanger"));

    // Creates ZK path for runnable
    zkClient.create("/" + runId.getId() + "/runnables", null, CreateMode.PERSISTENT).get();
    runningContainers.addWatcher(Constants.DISCOVERY_PATH_PREFIX);
    runnableContainerRequests = initContainerRequests();
  }

  @Override
  protected void doStop() throws Exception {
    Thread.interrupted();     // This is just to clear the interrupt flag

    LOG.info("Stop application master with spec: {}",
             TwillRuntimeSpecificationAdapter.create().toJson(twillRuntimeSpec));

    if (eventHandler != null) {
      try {
        // call event handler destroy. If there is error, only log and not affected stop sequence.
        eventHandler.destroy();
      } catch (Throwable t) {
        LOG.warn("Exception when calling {}.destroy()", eventHandler.getClass().getName(), t);
      }
    }

    instanceChangeExecutor.shutdownNow();

    // For checking if all containers are stopped.
    final Set<String> ids = Sets.newHashSet(runningContainers.getContainerIds());
    final YarnAMClient.AllocateHandler handler = new YarnAMClient.AllocateHandler() {
      @Override
      public void acquired(List<? extends ProcessLauncher<YarnContainerInfo>> launchers) {
        // no-op
      }

      @Override
      public void completed(List<YarnContainerStatus> completed) {
        for (YarnContainerStatus status : completed) {
          handleCompleted(completed);
          ids.remove(status.getContainerId());
        }
      }
    };

    // Handle heartbeats during shutdown because runningContainers.stopAll() waits until
    // handleCompleted() is called for every stopped runnable
    ExecutorService stopPoller = Executors.newSingleThreadExecutor(Threads.createDaemonThreadFactory("stopPoller"));
    stopPoller.execute(new Runnable() {
      @Override
      public void run() {
        while (!ids.isEmpty()) {
          try {
            amClient.allocate(0.0f, handler);
            if (!ids.isEmpty()) {
              TimeUnit.SECONDS.sleep(1);
            }
          } catch (Exception e) {
            LOG.error("Got exception while getting heartbeat", e);
          }
        }
      }
    });

    // runningContainers.stopAll() will wait for all the running runnables to stop or kill them after a timeout
    runningContainers.stopAll();
    // Since all the runnables are now stopped, it is okay to stop the poller.
    stopPoller.shutdownNow();
    cleanupDir();
  }

  @Override
  protected Object getLiveNodeData() {
    return amLiveNode;
  }

  @Override
  public ListenableFuture<String> onReceived(String messageId, Message message) {
    LOG.debug("Message received: {} {}.", messageId, message);

    SettableFuture<String> result = SettableFuture.create();
    Runnable completion = getMessageCompletion(messageId, result);

    if (handleSecureStoreUpdate(message)) {
      runningContainers.sendToAll(message, completion);
      return result;
    }

    if (handleSetInstances(message, completion)) {
      return result;
    }

    if (handleRestartRunnablesInstances(message, completion)) {
      return result;
    }

    // Replicate messages to all runnables
    if (message.getScope() == Message.Scope.ALL_RUNNABLE) {
      runningContainers.sendToAll(message, completion);
      return result;
    }

    // Replicate message to a particular runnable.
    if (message.getScope() == Message.Scope.RUNNABLE) {
      runningContainers.sendToRunnable(message.getRunnableName(), message, completion);
      return result;
    }

    LOG.info("Message ignored. {}", message);
    return Futures.immediateFuture(messageId);
  }

  @Override
  protected void triggerShutdown() {
    stopped = true;
  }

  private void cleanupDir() {
    try {
      if (applicationLocation.delete(true)) {
        LOG.info("Application directory deleted: {}", applicationLocation);
      } else {
        LOG.warn("Failed to cleanup directory {}.", applicationLocation);
      }
    } catch (Exception e) {
      LOG.warn("Exception while cleanup directory {}.", applicationLocation, e);
    }
  }

  @Override
  protected void doRun() throws Exception {
    // The main loop
    Map.Entry<AllocationSpecification, ? extends Collection<RuntimeSpecification>> currentRequest = null;
    final Queue<ProvisionRequest> provisioning = Lists.newLinkedList();

    YarnAMClient.AllocateHandler allocateHandler = new YarnAMClient.AllocateHandler() {
      @Override
      public void acquired(List<? extends ProcessLauncher<YarnContainerInfo>> launchers) {
        launchRunnable(launchers, provisioning);
      }

      @Override
      public void completed(List<YarnContainerStatus> completed) {
        handleCompleted(completed);
      }
    };

    long requestStartTime = 0;
    boolean isRequestRelaxed = false;
    long nextTimeoutCheck = System.currentTimeMillis() + Constants.PROVISION_TIMEOUT;
    while (!stopped) {
      // Call allocate. It has to be made at first in order to be able to get cluster resource availability.
      amClient.allocate(0.0f, allocateHandler);

      // Looks for containers requests.
      if (provisioning.isEmpty() && runnableContainerRequests.isEmpty() && runningContainers.isEmpty()) {
        LOG.info("All containers completed. Shutting down application master.");
        break;
      }

      // If nothing is in provisioning, and no pending request, move to next one
      int count = runnableContainerRequests.size();
      while (provisioning.isEmpty() && currentRequest == null && !runnableContainerRequests.isEmpty()) {
        RunnableContainerRequest runnableContainerRequest = runnableContainerRequests.peek();
        if (!runnableContainerRequest.isReadyToBeProvisioned()) {
          // take it out from queue and put it back at the end for second chance.
          runnableContainerRequest = runnableContainerRequests.poll();
          runnableContainerRequests.add(runnableContainerRequest);

          // We checked all the requests that were pending when we started this loop
          // Any remaining requests are not ready to be provisioned
          if (--count <= 0) {
            break;
          }
          continue;
        }
        currentRequest = runnableContainerRequest.takeRequest();
        if (currentRequest == null) {
          // All different types of resource request from current order is done, move to next one
          // TODO: Need to handle order type as well
          runnableContainerRequests.poll();
        }
      }

      // Nothing in provision, makes the next batch of provision request
      if (provisioning.isEmpty() && currentRequest != null) {
        manageBlacklist(currentRequest);
        addContainerRequests(currentRequest.getKey().getResource(), currentRequest.getValue(), provisioning,
                             currentRequest.getKey().getType());
        currentRequest = null;
        requestStartTime = System.currentTimeMillis();
        isRequestRelaxed = false;
      }

      // Check for provision request timeout i.e. check if any provision request has been pending
      // for more than the designated time. On timeout, relax the request constraints.
      if (!provisioning.isEmpty() && !isRequestRelaxed &&
        (System.currentTimeMillis() - requestStartTime) > Constants.CONSTRAINED_PROVISION_REQUEST_TIMEOUT) {
        LOG.info("Relaxing provisioning constraints for request {}", provisioning.peek().getRequestId());
        // Clear the blacklist for the pending provision request(s).
        amClient.clearBlacklist();
        isRequestRelaxed = true;
      }

      nextTimeoutCheck = checkProvisionTimeout(nextTimeoutCheck);

      if (isRunning()) {
        TimeUnit.SECONDS.sleep(1);
      }
    }
  }

  /**
   * Manage Blacklist for a given request.
   */
  private void manageBlacklist(Map.Entry<AllocationSpecification, ? extends Collection<RuntimeSpecification>> request) {
    amClient.clearBlacklist();

    //Check the allocation strategy
    AllocationSpecification currentAllocationSpecification = request.getKey();
    if (currentAllocationSpecification.getType().equals(AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME)) {

      //Check the placement policy
      TwillSpecification.PlacementPolicy placementPolicy =
        placementPolicyManager.getPlacementPolicy(currentAllocationSpecification.getRunnableName());
      if (placementPolicy.getType().equals(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED)) {

        //Update blacklist with hosts which are running DISTRIBUTED runnables
        for (String runnable : placementPolicyManager.getFellowRunnables(request.getKey().getRunnableName())) {
          Collection<ContainerInfo> containerStats =
            runningContainers.getContainerInfo(runnable);
          for (ContainerInfo containerInfo : containerStats) {
            // Yarn Resource Manager may include port in the node name depending on the setting
            // YarnConfiguration.RM_SCHEDULER_INCLUDE_PORT_IN_NODE_NAME. It is safe to add both
            // the names (with and without port) to the blacklist.
            amClient.addToBlacklist(containerInfo.getHost().getHostName());
            amClient.addToBlacklist(containerInfo.getHost().getHostName() + ":" + containerInfo.getPort());
          }
        }
      }
    }
  }

  /**
   * Handling containers that are completed.
   */
  private void handleCompleted(List<YarnContainerStatus> completedContainersStatuses) {
    Multiset<String> restartRunnables = HashMultiset.create();
    for (YarnContainerStatus status : completedContainersStatuses) {
      LOG.info("Container {} completed with {}:{}.",
               status.getContainerId(), status.getState(), status.getDiagnostics());
      runningContainers.handleCompleted(status, restartRunnables);
    }

    for (Multiset.Entry<String> entry : restartRunnables.entrySet()) {
      LOG.info("Re-request container for {} with {} instances.", entry.getElement(), entry.getCount());
      for (int i = 0; i < entry.getCount(); i++) {
        runnableContainerRequests.add(createRunnableContainerRequest(entry.getElement()));
      }
    }

    // For all runnables that needs to re-request for containers, update the expected count timestamp
    // so that the EventHandler would triggered with the right expiration timestamp.
    expectedContainers.updateRequestTime(restartRunnables.elementSet());
  }

  /**
   * Check for containers provision timeout and invoke eventHandler if necessary.
   *
   * @return the timestamp for the next time this method needs to be called.
   */
  private long checkProvisionTimeout(long nextTimeoutCheck) {
    if (System.currentTimeMillis() < nextTimeoutCheck) {
      return nextTimeoutCheck;
    }

    // Invoke event handler for provision request timeout
    Map<String, ExpectedContainers.ExpectedCount> expiredRequests = expectedContainers.getAll();
    Map<String, Integer> runningCounts = runningContainers.countAll();
    Map<String, Integer> completedContainerCount = runningContainers.getCompletedContainerCount();

    List<EventHandler.TimeoutEvent> timeoutEvents = Lists.newArrayList();
    for (Map.Entry<String, ExpectedContainers.ExpectedCount> entry : expiredRequests.entrySet()) {
      String runnableName = entry.getKey();
      ExpectedContainers.ExpectedCount expectedCount = entry.getValue();
      int runningCount = runningCounts.containsKey(runnableName) ? runningCounts.get(runnableName) : 0;
      int completedCount = completedContainerCount.containsKey(runnableName) ?
        completedContainerCount.get(runnableName) : 0;
      if (expectedCount.getCount() > runningCount + completedCount) {
        timeoutEvents.add(new EventHandler.TimeoutEvent(runnableName, expectedCount.getCount(),
                                                        runningCount, expectedCount.getTimestamp()));
      }
    }

    if (!timeoutEvents.isEmpty() && eventHandler != null) {
      try {
        EventHandler.TimeoutAction action = eventHandler.launchTimeout(timeoutEvents);
        if (action.getTimeout() < 0) {
          // Abort application
          stop();
        } else {
          return nextTimeoutCheck + action.getTimeout();
        }
      } catch (Throwable t) {
        LOG.warn("Exception when calling EventHandler {}. Ignore the result.", t);
      }
    }
    return nextTimeoutCheck + Constants.PROVISION_TIMEOUT;
  }

  private Credentials createCredentials() {
    Credentials credentials = new Credentials();
    if (!UserGroupInformation.isSecurityEnabled()) {
      return credentials;
    }

    try {
      credentials.addAll(UserGroupInformation.getCurrentUser().getCredentials());

      // Remove the AM->RM tokens
      Iterator<Token<?>> iter = credentials.getAllTokens().iterator();
      while (iter.hasNext()) {
        Token<?> token = iter.next();
        if (token.getKind().equals(AMRM_TOKEN_KIND_NAME)) {
          iter.remove();
        }
      }
    } catch (IOException e) {
      LOG.warn("Failed to get current user. No credentials will be provided to containers.", e);
    }

    return credentials;
  }

  private Queue<RunnableContainerRequest> initContainerRequests() {
    // Orderly stores container requests.
    Queue<RunnableContainerRequest> requests = new ConcurrentLinkedQueue<>();
    // For each order in the twillSpec, create container request for runnables, depending on Placement policy.
    for (TwillSpecification.Order order : twillSpec.getOrders()) {
      Set<String> distributedRunnables = placementPolicyManager.getDistributedRunnables(order.getNames());
      Set<String> defaultRunnables = Sets.newHashSet();
      defaultRunnables.addAll(order.getNames());
      defaultRunnables.removeAll(distributedRunnables);
      Map<AllocationSpecification, Collection<RuntimeSpecification>> requestsMap = Maps.newHashMap();
      for (String runnableName : distributedRunnables) {
        RuntimeSpecification runtimeSpec = twillSpec.getRunnables().get(runnableName);
        Resource capability = createCapability(runtimeSpec.getResourceSpecification());
        for (int instanceId = 0; instanceId < runtimeSpec.getResourceSpecification().getInstances(); instanceId++) {
          AllocationSpecification allocationSpecification =
            new AllocationSpecification(capability, AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME,
                                        runnableName, instanceId);
          addAllocationSpecification(allocationSpecification, requestsMap, runtimeSpec);
        }
      }
      for (String runnableName : defaultRunnables) {
        RuntimeSpecification runtimeSpec = twillSpec.getRunnables().get(runnableName);
        Resource capability = createCapability(runtimeSpec.getResourceSpecification());
        AllocationSpecification allocationSpecification = new AllocationSpecification(capability);
        addAllocationSpecification(allocationSpecification, requestsMap, runtimeSpec);
      }
      requests.add(new RunnableContainerRequest(order.getType(), requestsMap));
    }
    return requests;
  }

  /**
   * Helper method to create {@link org.apache.twill.internal.appmaster.RunnableContainerRequest}.
   */
  private void addAllocationSpecification(AllocationSpecification allocationSpecification,
                                          Map<AllocationSpecification, Collection<RuntimeSpecification>> map,
                                          RuntimeSpecification runtimeSpec) {
    if (!map.containsKey(allocationSpecification)) {
      map.put(allocationSpecification, Lists.<RuntimeSpecification>newLinkedList());
    }
    map.get(allocationSpecification).add(runtimeSpec);
  }

  /**
   * Adds container requests with the given resource capability for each runtime.
   */
  private void addContainerRequests(Resource capability,
                                    Collection<RuntimeSpecification> runtimeSpecs,
                                    Queue<ProvisionRequest> provisioning,
                                    AllocationSpecification.Type allocationType) {
    for (RuntimeSpecification runtimeSpec : runtimeSpecs) {
      String name = runtimeSpec.getName();
      int newContainers = expectedContainers.getExpected(name) - runningContainers.count(name);
      if (newContainers > 0) {
        if (allocationType.equals(AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME)) {
          //Spawning 1 instance at a time
          newContainers = 1;
        }
        TwillSpecification.PlacementPolicy placementPolicy = placementPolicyManager.getPlacementPolicy(name);
        Set<String> hosts = Sets.newHashSet();
        Set<String> racks = Sets.newHashSet();
        if (placementPolicy != null) {
          hosts = placementPolicy.getHosts();
          racks = placementPolicy.getRacks();
        }
        // TODO: Allow user to set priority?
        LOG.info("Request {} container with capability {} for runnable {}", newContainers, capability, name);
        String requestId = amClient.addContainerRequest(capability, newContainers)
          .addHosts(hosts)
          .addRacks(racks)
          .setPriority(0).apply();
        provisioning.add(new ProvisionRequest(runtimeSpec, requestId, newContainers, allocationType));
      }
    }
  }

  /**
   * Launches runnables in the provisioned containers.
   */
  private void launchRunnable(List<? extends ProcessLauncher<YarnContainerInfo>> launchers,
                              Queue<ProvisionRequest> provisioning) {
    for (ProcessLauncher<YarnContainerInfo> processLauncher : launchers) {
      LOG.info("Got container {}", processLauncher.getContainerInfo().getId());
      ProvisionRequest provisionRequest = provisioning.peek();
      if (provisionRequest == null) {
        continue;
      }

      String runnableName = provisionRequest.getRuntimeSpec().getName();
      LOG.info("Starting runnable {} with {}", runnableName, processLauncher);

      LOG.debug("Log level for Twill runnable {} is {}", runnableName, twillRuntimeSpec.getLogLevel());

      int containerCount = expectedContainers.getExpected(runnableName);

      // Setup container environment variables
      Map<String, String> env = new LinkedHashMap<>();
      if (environments.containsKey(runnableName)) {
        env.putAll(environments.get(runnableName));
      }
      // Override with system env
      env.put(EnvKeys.TWILL_LOG_KAFKA_ZK, getKafkaZKConnect());

      ProcessLauncher.PrepareLaunchContext launchContext = processLauncher.prepareLaunch(env, getLocalizeFiles(),
                                                                                         credentials);

      TwillContainerLauncher launcher = new TwillContainerLauncher(
        twillSpec.getRunnables().get(runnableName), processLauncher.getContainerInfo(), launchContext,
        ZKClients.namespace(zkClient, getZKNamespace(runnableName)),
        containerCount, jvmOpts, reservedMemory, getSecureStoreLocation());

      runningContainers.start(runnableName, processLauncher.getContainerInfo(), launcher,
                              twillRuntimeSpec.getLogLevel());

      // Need to call complete to workaround bug in YARN AMRMClient
      if (provisionRequest.containerAcquired()) {
        amClient.completeContainerRequest(provisionRequest.getRequestId());
      }

      if (expectedContainers.getExpected(runnableName) == runningContainers.count(runnableName) ||
        provisioning.peek().getType().equals(AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME)) {
        provisioning.poll();
      }
      if (expectedContainers.getExpected(runnableName) == runningContainers.count(runnableName)) {
        LOG.info("Runnable " + runnableName + " fully provisioned with " + containerCount + " instances.");
      }
    }
  }

  private List<LocalFile> getLocalizeFiles() {
    try (Reader reader = Files.newReader(new File(Constants.Files.LOCALIZE_FILES), Charsets.UTF_8)) {
      return new GsonBuilder().registerTypeAdapter(LocalFile.class, new LocalFileCodec())
        .create().fromJson(reader, new TypeToken<List<LocalFile>>() {
        }.getType());
    } catch (IOException e) {
      throw Throwables.propagate(e);
    }
  }

  private Map<String, Map<String, String>> getEnvironments() {
    File envFile = new File(Constants.Files.ENVIRONMENTS);
    if (!envFile.exists()) {
      return new HashMap<>();
    }

    try (Reader reader = Files.newReader(envFile, Charsets.UTF_8)) {
      return new Gson().fromJson(reader, new TypeToken<Map<String, Map<String, String>>>() {
      }.getType());
    } catch (IOException e) {
      throw Throwables.propagate(e);
    }
  }

  private String getZKNamespace(String runnableName) {
    return String.format("/%s/runnables/%s", runId.getId(), runnableName);
  }

  String getKafkaZKConnect() {
    return String.format("%s/%s/kafka", zkClient.getConnectString(), runId.getId());
  }

  /**
   * Attempts to change the number of running instances.
   *
   * @return {@code true} if the message does requests for changes in number of running instances of a runnable,
   * {@code false} otherwise.
   */
  private boolean handleSetInstances(final Message message, final Runnable completion) {
    if (message.getType() != Message.Type.SYSTEM || message.getScope() != Message.Scope.RUNNABLE) {
      return false;
    }

    Command command = message.getCommand();
    Map<String, String> options = command.getOptions();
    if (!"instances".equals(command.getCommand()) || !options.containsKey("count")) {
      return false;
    }

    final String runnableName = message.getRunnableName();
    if (runnableName == null || runnableName.isEmpty() || !twillSpec.getRunnables().containsKey(runnableName)) {
      LOG.info("Unknown runnable {}", runnableName);
      return false;
    }

    final int newCount = Integer.parseInt(options.get("count"));
    final int oldCount = expectedContainers.getExpected(runnableName);

    LOG.info("Received change instances request for {}, from {} to {}.", runnableName, oldCount, newCount);

    if (newCount == oldCount) {   // Nothing to do, simply complete the request.
      completion.run();
      return true;
    }

    instanceChangeExecutor.execute(createSetInstanceRunnable(message, completion, oldCount, newCount));
    return true;
  }

  /**
   * Creates a Runnable for execution of change instance request.
   */
  private Runnable createSetInstanceRunnable(final Message message, final Runnable completion,
                                             final int oldCount, final int newCount) {
    return new Runnable() {
      @Override
      public void run() {
        final String runnableName = message.getRunnableName();

        LOG.info("Processing change instance request for {}, from {} to {}.", runnableName, oldCount, newCount);
        try {
          // Wait until running container count is the same as old count
          runningContainers.waitForCount(runnableName, oldCount);
          LOG.info("Confirmed {} containers running for {}.", oldCount, runnableName);

          expectedContainers.setExpected(runnableName, newCount);

          try {
            if (newCount < oldCount) {
              // Shutdown some running containers
              for (int i = 0; i < oldCount - newCount; i++) {
                runningContainers.stopLastAndWait(runnableName);
              }
            } else {
              // Increase the number of instances
              runnableContainerRequests.add(createRunnableContainerRequest(runnableName, newCount - oldCount));
            }
          } finally {
            // Send a message to all running runnables that number of instances have changed
            runningContainers.sendToRunnable(runnableName, message, completion);
            LOG.info("Change instances request completed. From {} to {}.", oldCount, newCount);
          }
        } catch (InterruptedException e) {
          // If the wait is being interrupted, discard the message.
          completion.run();
        }
      }
    };
  }

  private RunnableContainerRequest createRunnableContainerRequest(final String runnableName) {
    return createRunnableContainerRequest(runnableName, 1);
  }

  private RunnableContainerRequest createRunnableContainerRequest(final String runnableName,
                                                                  final int numberOfInstances) {
    return createRunnableContainerRequest(runnableName, numberOfInstances, true);
  }

  private RunnableContainerRequest createRunnableContainerRequest(final String runnableName,
                                                                  final int numberOfInstances,
                                                                  final boolean isProvisioned) {
    // Find the current order of the given runnable in order to create a RunnableContainerRequest.
    TwillSpecification.Order order = Iterables.find(twillSpec.getOrders(), new Predicate<TwillSpecification.Order>() {
      @Override
      public boolean apply(TwillSpecification.Order input) {
        return (input.getNames().contains(runnableName));
      }
    });

    RuntimeSpecification runtimeSpec = twillSpec.getRunnables().get(runnableName);
    Resource capability = createCapability(runtimeSpec.getResourceSpecification());
    Map<AllocationSpecification, Collection<RuntimeSpecification>> requestsMap = Maps.newHashMap();
    if (placementPolicyManager.getPlacementPolicyType(runnableName).equals(
      TwillSpecification.PlacementPolicy.Type.DISTRIBUTED)) {
      for (int instanceId = 0; instanceId < numberOfInstances; instanceId++) {
        AllocationSpecification allocationSpecification =
          new AllocationSpecification(capability, AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME,
                                      runnableName, instanceId);
        addAllocationSpecification(allocationSpecification, requestsMap, runtimeSpec);
      }
    } else {
      AllocationSpecification allocationSpecification = new AllocationSpecification(capability);
      addAllocationSpecification(allocationSpecification, requestsMap, runtimeSpec);
    }
    return new RunnableContainerRequest(order.getType(), requestsMap, isProvisioned);
  }

  private Runnable getMessageCompletion(final String messageId, final SettableFuture<String> future) {
    return new Runnable() {
      @Override
      public void run() {
        future.set(messageId);
      }
    };
  }

  private Resource createCapability(ResourceSpecification resourceSpec) {
    Resource capability = Records.newRecord(Resource.class);

    if (!YarnUtils.setVirtualCores(capability, resourceSpec.getVirtualCores())) {
      LOG.debug("Virtual cores limit not supported.");
    }

    capability.setMemory(resourceSpec.getMemorySize());
    return capability;
  }

  /**
   * Attempt to restart some instances from a runnable or some runnables.
   *
   * @return {@code true} if the message requests restarting some instances and {@code false} otherwise.
   */
  private boolean handleRestartRunnablesInstances(final Message message, final Runnable completion) {
    LOG.debug("Check if it should process a restart runnable instances.");

    if (message.getType() != Message.Type.SYSTEM) {
      return false;
    }

    Message.Scope messageScope = message.getScope();
    if (messageScope != Message.Scope.RUNNABLE && messageScope != Message.Scope.RUNNABLES) {
      return false;
    }

    Command requestCommand = message.getCommand();
    if (!Constants.RESTART_ALL_RUNNABLE_INSTANCES.equals(requestCommand.getCommand()) &&
      !Constants.RESTART_RUNNABLES_INSTANCES.equals(requestCommand.getCommand())) {
      return false;
    }

    LOG.debug("Processing restart runnable instances message {}.", message);

    if (!Strings.isNullOrEmpty(message.getRunnableName()) && message.getScope() == Message.Scope.RUNNABLE) {
      // ... for a runnable ...
      String runnableName = message.getRunnableName();
      LOG.debug("Start restarting all runnable {} instances.", runnableName);
      restartRunnableInstances(runnableName, null, completion);
    } else {
      // ... or maybe some runnables
      for (Map.Entry<String, String> option : requestCommand.getOptions().entrySet()) {
        String runnableName = option.getKey();
        Set<Integer> restartedInstanceIds = GSON.fromJson(option.getValue(),
                                                          new TypeToken<Set<Integer>>() {
                                                          }.getType());

        LOG.debug("Start restarting runnable {} instances {}", runnableName, restartedInstanceIds);
        restartRunnableInstances(runnableName, restartedInstanceIds, completion);
      }
    }

    return true;
  }

  /**
   * Helper method to restart instances of runnables.
   */
  private void restartRunnableInstances(String runnableName, @Nullable Set<Integer> instanceIds,
                                        final Runnable completion) {
    Runnable restartInstancesRunnable = createRestartInstancesRunnable(runnableName, instanceIds, completion);
    instanceChangeExecutor.execute(restartInstancesRunnable);
  }

  /**
   * Creates a Runnable for execution of restart instances request.
   */
  private Runnable createRestartInstancesRunnable(final String runnableName, @Nullable final Set<Integer> instanceIds,
                                                  final Runnable completion) {
    return new Runnable() {
      @Override
      public void run() {
        LOG.debug("Begin restart runnable {} instances.", runnableName);
        int runningCount = runningContainers.count(runnableName);
        Set<Integer> instancesToRemove = instanceIds == null ? null : ImmutableSet.copyOf(instanceIds);
        if (instancesToRemove == null) {
          instancesToRemove = Ranges.closedOpen(0, runningCount).asSet(DiscreteDomains.integers());
        }

        LOG.info("Restarting instances {} for runnable {}", instancesToRemove, runnableName);
        RunnableContainerRequest containerRequest =
          createRunnableContainerRequest(runnableName, instancesToRemove.size(), false);
        runnableContainerRequests.add(containerRequest);

        for (int instanceId : instancesToRemove) {
          LOG.debug("Stop instance {} for runnable {}", instanceId, runnableName);
          try {
            runningContainers.stopByIdAndWait(runnableName, instanceId);
          } catch (Exception ex) {
            // could be thrown if the container already stopped.
            LOG.info("Exception thrown when stopping instance {} probably already stopped.", instanceId);
          }
        }

        // set the container request to be ready
        containerRequest.setReadyToBeProvisioned(true);

        // For all runnables that needs to re-request for containers, update the expected count timestamp
        // so that the EventHandler would be triggered with the right expiration timestamp.
        expectedContainers.updateRequestTime(Collections.singleton(runnableName));

        completion.run();
      }
    };
  }
}