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

import java.io.IOException;
import java.lang.Thread.UncaughtExceptionHandler;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.AMRMProtocol;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ContainerManager;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
import org.apache.hadoop.yarn.api.records.AMResponse;
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.ContainerState;
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.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.s4.deploy.HdfsFetcherModule;
import org.apache.s4.tools.Tools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;

/**
 * An ApplicationMaster for launching S4 nodes on a set of launched containers using the YARN framework.
 * 
 * The code is inspired by the shell example from {@link http
 * ://hadoop.apache.org/docs/r2.0.2-alpha/hadoop-yarn/hadoop-yarn-site/WritingYarnApplications.html}
 * 
 * 
 */
public class S4ApplicationMaster {

    private static Logger logger = LoggerFactory.getLogger(S4ApplicationMaster.class);

    // Configuration
    private Configuration conf;

    // YARN RPC to communicate with the Resource Manager or Node Manager
    private YarnRPC rpc;

    // Handle to communicate with the Resource Manager
    private AMRMProtocol resourceManager;

    // Application Attempt Id ( combination of attemptId and fail count )
    private ApplicationAttemptId appAttemptID;

    // For status update for clients - yet to be implemented
    // Hostname of the container
    private final String appMasterHostname = "";
    // Port on which the app master listens for status update requests from clients
    private final int appMasterRpcPort = 0;
    // Tracking url to which app master publishes info for clients to monitor
    private final String appMasterTrackingUrl = "";

    // Incremental counter for rpc calls to the RM
    private final AtomicInteger rmRequestID = new AtomicInteger();

    // Simple flag to denote whether all works is done
    private boolean appDone = false;
    // Counter for completed containers ( complete denotes successful or failed )
    private final AtomicInteger numCompletedContainers = new AtomicInteger();
    // Allocated container count so that we know how many containers has the RM
    // allocated to us
    private final AtomicInteger numAllocatedContainers = new AtomicInteger();
    // Count of failed containers
    private final AtomicInteger numFailedContainers = new AtomicInteger();
    // Count of containers already requested from the RM
    // Needed as once requested, we should not request for containers again and again.
    // Only request for more if the original requirement changes.
    private final AtomicInteger numRequestedContainers = new AtomicInteger();

    // Containers to be released
    private final CopyOnWriteArrayList<ContainerId> releasedContainers = new CopyOnWriteArrayList<ContainerId>();

    // Launch threads
    private final List<Thread> launchThreads = new ArrayList<Thread>();

    private int containerMemory;

    private static AppMasterYarnArgs yarnArgs;

    /**
     * @param args
     *            Command line args
     */
    public static void main(String[] args) {
        boolean result = false;
        try {
            yarnArgs = new AppMasterYarnArgs();
            logger.info("S4YarnApplicationMaster args = " + Arrays.toString(args));

            Tools.parseArgs(yarnArgs, args);

            S4ApplicationMaster appMaster = new S4ApplicationMaster();
            logger.info("Initializing ApplicationMaster with args " + Arrays.toString(args));
            appMaster.init();
            result = appMaster.run();
        } catch (Throwable t) {
            t.printStackTrace();
            logger.error("Error running ApplicationMaster", t);
            System.exit(1);
        }
        if (result) {
            logger.info("Application Master completed successfully. exiting");
            System.exit(0);
        } else {
            logger.info("Application Master failed. exiting");
            System.exit(2);
        }
    }

    public S4ApplicationMaster() throws Exception {

        Thread.setDefaultUncaughtExceptionHandler(new UncaughtExceptionHandler() {

            @Override
            public void uncaughtException(Thread t, Throwable e) {
                logger.error("Uncaught exception in thread {}", t.getName(), e);

            }
        });
    }

    /**
     */
    public void init() throws IOException {

        containerMemory = Utils.extractMemoryParam(yarnArgs.containerMemory, yarnArgs.extraS4NodeJVMParams);

        Map<String, String> envs = System.getenv();

        appAttemptID = Records.newRecord(ApplicationAttemptId.class);
        if (!envs.containsKey(ApplicationConstants.AM_CONTAINER_ID_ENV)) {
            if (!Strings.isNullOrEmpty(yarnArgs.appAttemptId)) {
                appAttemptID = ConverterUtils.toApplicationAttemptId(yarnArgs.appAttemptId);
            } else {
                throw new IllegalArgumentException("Application Attempt Id not set in the environment");
            }
        } else {
            ContainerId containerId = ConverterUtils.toContainerId(envs.get(ApplicationConstants.AM_CONTAINER_ID_ENV));
            appAttemptID = containerId.getApplicationAttemptId();
        }

        logger.info("Application master for app" + ", appId=" + appAttemptID.getApplicationId().getId()
                + ", clustertimestamp=" + appAttemptID.getApplicationId().getClusterTimestamp() + ", attemptId="
                + appAttemptID.getAttemptId());

        if (Strings.isNullOrEmpty(yarnArgs.cluster)) {
            throw new IllegalArgumentException("No cluster ID specified to be provisioned by application master");
        }

        conf = new YarnConfiguration();
        if (yarnArgs.test) {
            // testMode = true;
            conf.set(FileSystem.FS_DEFAULT_NAME_KEY, "hdfs://localhost:9000");
        }
        rpc = YarnRPC.create(conf);
    }

    /**
     * Main run function for the application master
     * 
     * @throws YarnRemoteException
     */
    public boolean run() throws YarnRemoteException {
        logger.info("Starting ApplicationMaster");

        // Connect to ResourceManager
        resourceManager = connectToRM();

        // Setup local RPC Server to accept status requests directly from clients
        // TODO need to setup a protocol for client to be able to communicate to the RPC server
        // TODO use the rpc port info to register with the RM for the client to send requests to this app master

        // Register self with ResourceManager
        RegisterApplicationMasterResponse response = registerToRM();
        // Dump out information about cluster capability as seen by the resource manager
        int minMem = response.getMinimumResourceCapability().getMemory();
        int maxMem = response.getMaximumResourceCapability().getMemory();
        logger.info("Min mem capability of resources in this cluster " + minMem);
        logger.info("Max mem capability of resources in this cluster " + maxMem);

        // A resource ask has to be atleast the minimum of the capability of the cluster, the value has to be
        // a multiple of the min value and cannot exceed the max.
        // If it is not an exact multiple of min, the RM will allocate to the nearest multiple of min
        if (containerMemory < minMem) {
            logger.info("Container memory for S4 node specified below min threshold of YARN cluster. Using min value."
                    + ", specified=" + containerMemory + ", min=" + minMem);
            containerMemory = minMem;
        } else if (containerMemory > maxMem) {
            logger.info("Container memory for S4 node specified above max threshold of YARN cluster. Using max value."
                    + ", specified=" + containerMemory + ", max=" + maxMem);
            containerMemory = maxMem;
        }

        int loopCounter = -1;

        while (numCompletedContainers.get() < yarnArgs.numContainers && !appDone) {
            loopCounter++;

            // log current state
            logger.info("Current application state: loop=" + loopCounter + ", appDone=" + appDone + ", total="
                    + yarnArgs.numContainers + ", requested=" + numRequestedContainers + ", completed="
                    + numCompletedContainers + ", failed=" + numFailedContainers + ", currentAllocated="
                    + numAllocatedContainers);

            // Sleep before each loop when asking RM for containers
            // to avoid flooding RM with spurious requests when it
            // need not have any available containers
            // Sleeping for 1000 ms.
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                logger.info("Sleep interrupted " + e.getMessage());
            }

            // No. of containers to request
            // For the first loop, askCount will be equal to total containers needed
            // From that point on, askCount will always be 0 as current implementation
            // does not change its ask on container failures.
            int askCount = yarnArgs.numContainers - numRequestedContainers.get();
            numRequestedContainers.addAndGet(askCount);

            // Setup request to be sent to RM to allocate containers
            List<ResourceRequest> resourceReq = new ArrayList<ResourceRequest>();
            if (askCount > 0) {
                ResourceRequest containerAsk = setupContainerAskForRM(askCount);
                resourceReq.add(containerAsk);
            }

            // Send the request to RM
            logger.info("Asking RM for containers" + ", askCount=" + askCount);
            AMResponse amResp = sendContainerAskToRM(resourceReq);

            // Retrieve list of allocated containers from the response
            List<Container> allocatedContainers = amResp.getAllocatedContainers();
            logger.info("Got response from RM for container ask, allocatedCnt=" + allocatedContainers.size());
            numAllocatedContainers.addAndGet(allocatedContainers.size());
            for (Container allocatedContainer : allocatedContainers) {
                logger.info("Launching S4 node on a new container." + ", containerId=" + allocatedContainer.getId()
                        + ", containerNode=" + allocatedContainer.getNodeId().getHost() + ":"
                        + allocatedContainer.getNodeId().getPort() + ", containerNodeURI="
                        + allocatedContainer.getNodeHttpAddress() + ", containerState" + allocatedContainer.getState()
                        + ", containerResourceMemory" + allocatedContainer.getResource().getMemory());
                // + ", containerToken" + allocatedContainer.getContainerToken().getIdentifier().toString());

                LaunchContainerRunnable runnableLaunchContainer = new LaunchContainerRunnable(allocatedContainer);
                Thread launchThread = new Thread(runnableLaunchContainer);

                // 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.
                launchThreads.add(launchThread);
                launchThread.start();
            }

            // Check what the current available resources in the cluster are
            // TODO should we do anything if the available resources are not enough?
            Resource availableResources = amResp.getAvailableResources();
            logger.info("Current available resources in the cluster " + availableResources);

            // Check the completed containers
            List<ContainerStatus> completedContainers = amResp.getCompletedContainersStatuses();
            logger.info("Got response from RM for container ask, completedCnt=" + completedContainers.size());
            for (ContainerStatus containerStatus : completedContainers) {
                logger.info("Got container status for containerID= " + containerStatus.getContainerId() + ", state="
                        + containerStatus.getState() + ", exitStatus=" + containerStatus.getExitStatus()
                        + ", diagnostics=" + containerStatus.getDiagnostics());

                // non complete containers should not be here
                assert (containerStatus.getState() == ContainerState.COMPLETE);

                // increment counters for completed/failed containers
                int exitStatus = containerStatus.getExitStatus();
                if (0 != exitStatus) {
                    // container failed
                    if (-100 != exitStatus) {
                        // shell script failed
                        // counts as completed
                        numCompletedContainers.incrementAndGet();
                        numFailedContainers.incrementAndGet();
                    } else {
                        // something else bad happened
                        // app job did not complete for some reason
                        // we should re-try as the container was lost for some reason
                        numAllocatedContainers.decrementAndGet();
                        numRequestedContainers.decrementAndGet();
                        // we do not need to release the container as it would be done
                        // by the RM/CM.
                    }
                } else {
                    // nothing to do
                    // container completed successfully
                    numCompletedContainers.incrementAndGet();
                    logger.info("Container completed successfully." + ", containerId="
                            + containerStatus.getContainerId());
                }

            }
            if (numCompletedContainers.get() == yarnArgs.numContainers) {
                appDone = true;
            }

            logger.info("Current application state: loop=" + loopCounter + ", appDone=" + appDone + ", total="
                    + yarnArgs.numContainers + ", requested=" + numRequestedContainers + ", completed="
                    + numCompletedContainers + ", failed=" + numFailedContainers + ", currentAllocated="
                    + numAllocatedContainers);

        }

        // Join all launched threads
        // needed for when we time out
        // and we need to release containers
        for (Thread launchThread : launchThreads) {
            try {
                launchThread.join(0);
            } catch (InterruptedException e) {
                logger.info("Exception thrown in thread join: " + e.getMessage());
                e.printStackTrace();
            }
        }

        // When the application completes, it should send a finish application signal
        // to the RM
        logger.info("Application completed. Signalling finish to RM");

        FinishApplicationMasterRequest finishReq = Records.newRecord(FinishApplicationMasterRequest.class);
        finishReq.setAppAttemptId(appAttemptID);
        boolean isSuccess = true;
        if (numFailedContainers.get() == 0) {
            finishReq.setFinishApplicationStatus(FinalApplicationStatus.SUCCEEDED);
        } else {
            finishReq.setFinishApplicationStatus(FinalApplicationStatus.FAILED);
            String diagnostics = "Diagnostics." + ", total=" + yarnArgs.numContainers + ", completed="
                    + numCompletedContainers.get() + ", allocated=" + numAllocatedContainers.get() + ", failed="
                    + numFailedContainers.get();
            finishReq.setDiagnostics(diagnostics);
            isSuccess = false;
        }
        resourceManager.finishApplicationMaster(finishReq);
        return isSuccess;
    }

    /**
     * Thread to connect to the {@link ContainerManager} and launch the container that will execute the shell command.
     */
    private class LaunchContainerRunnable implements Runnable {

        // Allocated container
        Container container;
        // Handle to communicate with ContainerManager
        ContainerManager cm;

        /**
         * @param lcontainer
         *            Allocated container
         */
        public LaunchContainerRunnable(Container lcontainer) {
            this.container = lcontainer;
        }

        /**
         * Helper function to connect to CM
         */
        private void connectToCM() {
            logger.debug("Connecting to ContainerManager for containerid=" + container.getId());
            String cmIpPortStr = container.getNodeId().getHost() + ":" + container.getNodeId().getPort();
            InetSocketAddress cmAddress = NetUtils.createSocketAddr(cmIpPortStr);
            logger.info("Connecting to ContainerManager at " + cmIpPortStr);
            this.cm = ((ContainerManager) rpc.getProxy(ContainerManager.class, cmAddress, conf));
        }

        @Override
        /**
         * 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
            connectToCM();

            logger.info("Setting up container launch container for containerid=" + container.getId());
            ContainerLaunchContext ctx = Records.newRecord(ContainerLaunchContext.class);

            ctx.setContainerId(container.getId());
            ctx.setResource(container.getResource());

            try {
                if (!Strings.isNullOrEmpty(yarnArgs.user)) {
                    ctx.setUser(yarnArgs.user);
                } else {
                    logger.info("Using default user name {}", UserGroupInformation.getCurrentUser().getShortUserName());
                    ctx.setUser(UserGroupInformation.getCurrentUser().getShortUserName());
                }
            } catch (IOException e) {
                logger.info("Getting current user info failed when trying to launch the container" + e.getMessage());
            }

            // Set the local resources
            Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();

            try {
                FileSystem fs = FileSystem.get(conf);

                RemoteIterator<LocatedFileStatus> files = fs.listFiles(new Path(fs.getHomeDirectory(), "/app-"
                        + appAttemptID.getApplicationId().getId()), false);
                while (files.hasNext()) {
                    LocatedFileStatus file = files.next();
                    LocalResource localResource = Records.newRecord(LocalResource.class);

                    localResource.setType(LocalResourceType.FILE);
                    localResource.setVisibility(LocalResourceVisibility.APPLICATION);
                    localResource.setResource(ConverterUtils.getYarnUrlFromPath(file.getPath()));
                    localResource.setTimestamp(file.getModificationTime());
                    localResource.setSize(file.getLen());
                    localResources.put(file.getPath().getName(), localResource);
                }
                ctx.setLocalResources(localResources);

            } catch (IOException e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            }

            StringBuilder classPathEnv = new StringBuilder("${CLASSPATH}:./*");

            for (String c : conf.getStrings(YarnConfiguration.YARN_APPLICATION_CLASSPATH,
                    YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) {
                classPathEnv.append(':');
                classPathEnv.append(c.trim());
            }

            // classPathEnv.append(System.getProperty("java.class.path"));
            Map<String, String> env = new HashMap<String, String>();

            env.put("CLASSPATH", classPathEnv.toString());
            ctx.setEnvironment(env);

            // Set the necessary command to execute on the allocated container
            Vector<CharSequence> vargs = new Vector<CharSequence>(5);

            vargs.add("java");

            vargs.add("-Xmx" + containerMemory + "m");

            S4YarnClient.addListElementsToCommandLineBuffer(vargs, null, " ", yarnArgs.extraS4NodeJVMParams);

            // TODO add memory parameter
            // vargs.add("-Xdebug");
            // vargs.add("-Xrunjdwp:transport=dt_socket,address=8889,server=y");
            vargs.add("org.apache.s4.core.Main");
            vargs.add("-zk=" + yarnArgs.zkString);
            vargs.add("-c=" + yarnArgs.cluster);

            List<String> extraModulesClasses = Lists.newArrayList(yarnArgs.extraModulesClasses);
            // add module for fetchings from hdfs
            extraModulesClasses.add(HdfsFetcherModule.class.getName());
            S4YarnClient.addListElementsToCommandLineBuffer(vargs, CommonS4YarnArgs.EXTRA_MODULES_CLASSES, ",",
                    extraModulesClasses);

            // add reference to the configuration
            List<String> namedStringParams = Lists.newArrayList(yarnArgs.extraNamedParameters);
            if (yarnArgs.test) {
                namedStringParams.add(FileSystem.FS_DEFAULT_NAME_KEY + "=" + conf.get(FileSystem.FS_DEFAULT_NAME_KEY));
            }
            S4YarnClient.addListElementsToCommandLineBuffer(vargs, CommonS4YarnArgs.NAMED_STRING_PARAMETERS, ",",
                    namedStringParams);

            // TODO
            // We should redirect the output to hdfs instead of local logs
            // so as to be able to look at the final output after the containers
            // have been released.
            // Could use a path suffixed with /AppId/AppAttempId/ContainerId/std[out|err]
            vargs.add("1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/s4-node-stdout");
            vargs.add("2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/s4-node-stderr");

            // Get final commmand
            StringBuilder command = new StringBuilder();
            for (CharSequence str : vargs) {
                command.append(str).append(" ");
            }

            List<String> commands = new ArrayList<String>();
            commands.add(command.toString());
            ctx.setCommands(commands);

            StartContainerRequest startReq = Records.newRecord(StartContainerRequest.class);
            startReq.setContainerLaunchContext(ctx);
            try {
                cm.startContainer(startReq);
            } catch (YarnRemoteException e) {
                logger.info("Start container failed for :" + ", containerId=" + container.getId());
                e.printStackTrace();
            }

        }
    }

    /**
     * Connect to the Resource Manager
     * 
     * @return Handle to communicate with the RM
     */
    private AMRMProtocol connectToRM() {
        YarnConfiguration yarnConf = new YarnConfiguration(conf);
        InetSocketAddress rmAddress = yarnConf.getSocketAddr(YarnConfiguration.RM_SCHEDULER_ADDRESS,
                YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS, YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
        logger.info("Connecting to ResourceManager at " + rmAddress);
        return ((AMRMProtocol) rpc.getProxy(AMRMProtocol.class, rmAddress, conf));
    }

    /**
     * Register the Application Master to the Resource Manager
     * 
     * @return the registration response from the RM
     * @throws YarnRemoteException
     */
    private RegisterApplicationMasterResponse registerToRM() throws YarnRemoteException {
        RegisterApplicationMasterRequest appMasterRequest = Records.newRecord(RegisterApplicationMasterRequest.class);

        // set the required info into the registration request:
        // application attempt id,
        // host on which the app master is running
        // rpc port on which the app master accepts requests from the client
        // tracking url for the app master
        appMasterRequest.setApplicationAttemptId(appAttemptID);
        appMasterRequest.setHost(appMasterHostname);
        appMasterRequest.setRpcPort(appMasterRpcPort);
        appMasterRequest.setTrackingUrl(appMasterTrackingUrl);

        return resourceManager.registerApplicationMaster(appMasterRequest);
    }

    /**
     * Setup the request that will be sent to the RM for the container ask.
     * 
     * @param numContainers
     *            Containers to ask for from RM
     * @return the setup ResourceRequest to be sent to RM
     */
    private ResourceRequest setupContainerAskForRM(int numContainers) {
        ResourceRequest request = Records.newRecord(ResourceRequest.class);

        // setup requirements for hosts
        // whether a particular rack/host is needed
        // Refer to apis under org.apache.hadoop.net for more
        // details on how to get figure out rack/host mapping.
        // using * as any host will do for the distributed shell app
        request.setHostName("*");

        // set no. of containers needed
        request.setNumContainers(numContainers);

        // set the priority for the request
        Priority pri = Records.newRecord(Priority.class);
        // TODO - what is the range for priority? how to decide?
        pri.setPriority(yarnArgs.priority);
        request.setPriority(pri);

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

        return request;
    }

    /**
     * Ask RM to allocate given no. of containers to this Application Master
     * 
     * @param requestedContainers
     *            Containers to ask for from RM
     * @return Response from RM to AM with allocated containers
     * @throws YarnRemoteException
     */
    private AMResponse sendContainerAskToRM(List<ResourceRequest> requestedContainers) throws YarnRemoteException {
        AllocateRequest req = Records.newRecord(AllocateRequest.class);
        req.setResponseId(rmRequestID.incrementAndGet());
        req.setApplicationAttemptId(appAttemptID);
        req.addAllAsks(requestedContainers);
        req.addAllReleases(releasedContainers);
        req.setProgress((float) numCompletedContainers.get() / yarnArgs.numContainers);

        logger.info("Sending request to RM for containers" + ", requestedSet=" + requestedContainers.size()
                + ", releasedSet=" + releasedContainers.size() + ", progress=" + req.getProgress());

        for (ResourceRequest rsrcReq : requestedContainers) {
            logger.info("Requested container ask: " + rsrcReq.toString());
        }
        for (ContainerId id : releasedContainers) {
            logger.info("Released container, id=" + id.getId());
        }

        AllocateResponse resp = resourceManager.allocate(req);
        return resp.getAMResponse();
    }
}