@Override
    public void launchContainers() throws Exception {
        String noOfContainers = getJobConfiguration().getProperty(SingularConstants.KEY_NO_OF_CONTAINERS);
        String containerClassName = getJobConfiguration().getProperty(SingularConstants.KEY_CONTAINER_CLASS);
        String containerCores = getJobConfiguration().getProperty(SingularConstants.KEY_NO_OF_CORES);
        String containerMemory = getJobConfiguration().getProperty(SingularConstants.KEY_MEM);
        String stagingPath = getJobConfiguration().getProperty(SingularConstants.KEY_APPLICATION_STAGING_PATH);

        System.out.println("No of container " + noOfContainers + " " + containerClassName + " " +
        containerCores + " " + containerMemory);

        int noOfContainersRequired = Integer.parseInt(noOfContainers);
        int numberOfCompletedContainers = 0;
        int noOfContainersReceived = 0;
        int totalNoOfContainersReceived = 0;
        int noOfContainersToAsk = noOfContainersRequired;
        List<ContainerId> releasedContainers = new ArrayList<ContainerId>();

        while(totalNoOfContainersReceived < noOfContainersRequired || numberOfCompletedContainers < noOfContainersRequired) {
            //allocate would also act as heart beat to resource manager as well.
            AllocateResponse allocateResponse = resourceAllocator.allocateContainers(noOfContainersToAsk,releasedContainers,
                                                                                              numberOfCompletedContainers);

            List<ContainerStatus> statuses = allocateResponse.getAMResponse().getCompletedContainersStatuses();
            for(ContainerStatus status : statuses) {
                if(status.getState() == ContainerState.COMPLETE)
                    numberOfCompletedContainers++;
            }
            List<Container> containers = allocateResponse.getAMResponse().getAllocatedContainers();
            noOfContainersReceived = containers.size();
            totalNoOfContainersReceived += noOfContainersReceived;

            for(Container container : containers) {
                Runnable runnable = new ContainerRunnable(container,containerClassName,new Path(stagingPath));
                executorService.submit(runnable);
            }
            releasedContainers.clear();

            for(ContainerStatus containerStatus : allocateResponse.getAMResponse().getCompletedContainersStatuses()) {
                ContainerId containerId = containerStatus.getContainerId();
                releasedContainers.add(containerId);
            }

            System.out.println("Requested for " + noOfContainersToAsk + " , got " + noOfContainersReceived + " and no of completed containers " + numberOfCompletedContainers);

            // not to hit RM is quick succession.
            TimeUnit.SECONDS.sleep(2);
            noOfContainersToAsk = (noOfContainersRequired - totalNoOfContainersReceived);
        }

        executorService.shutdown();

        // waiting for the executor to finish all the threads.
        while (!executorService.isTerminated()) {
            TimeUnit.SECONDS.sleep(2);
        }
    }