@Override
  public void run() {
    ContainerId containerId = container.getContainerId();
    String containerIdStr = containerId.toString();
    LOG.info("Cleaning up container " + containerIdStr);

    try {
      context.getNMStateStore().storeContainerKilled(containerId);
    } catch (IOException e) {
      LOG.error("Unable to mark container " + containerId
          + " killed in store", e);
    }

    // launch flag will be set to true if process already launched,
    // in process of launching, or failed to launch.
    boolean alreadyLaunched = !launch.markLaunched() ||
        launch.isLaunchCompleted();
    if (!alreadyLaunched) {
      LOG.info("Container " + containerIdStr + " not launched."
          + " No cleanup needed to be done");
      return;
    }
    LOG.debug("Marking container {} as inactive", containerIdStr);
    // this should ensure that if the container process has not launched
    // by this time, it will never be launched
    exec.deactivateContainer(containerId);
    Path pidFilePath = launch.getPidFilePath();
    LOG.debug("Getting pid for container {} to kill"
        + " from pid file {}", containerIdStr, pidFilePath != null ?
        pidFilePath : "null");
    // however the container process may have already started
    try {

      // get process id from pid file if available
      // else if shell is still active, get it from the shell
      String processId = launch.getContainerPid();

      // kill process
      String user = container.getUser();
      if (processId != null) {
        signalProcess(processId, user, containerIdStr);
      } else {
        // Normally this means that the process was notified about
        // deactivateContainer above and did not start.
        // Since we already set the state to RUNNING or REINITIALIZING
        // we have to send a killed event to continue.
        if (!launch.isLaunchCompleted()) {
          LOG.warn("Container clean up before pid file created "
              + containerIdStr);
          dispatcher.getEventHandler().handle(
              new ContainerExitEvent(container.getContainerId(),
                  ContainerEventType.CONTAINER_KILLED_ON_REQUEST,
                  Shell.WINDOWS ?
                      ContainerExecutor.ExitCode.FORCE_KILLED.getExitCode() :
                      ContainerExecutor.ExitCode.TERMINATED.getExitCode(),
                  "Container terminated before pid file created."));
          // There is a possibility that the launch grabbed the file name before
          // the deactivateContainer above but it was slow enough to avoid
          // getContainerPid.
          // Increasing YarnConfiguration.NM_PROCESS_KILL_WAIT_MS
          // reduces the likelihood of this race condition and process leak.
        }
      }

      // rm container in docker
      if (DockerLinuxContainerRuntime.isDockerContainerRequested(conf,
          container.getLaunchContext().getEnvironment())) {
        rmDockerContainerDelayed();
      }
    } catch (Exception e) {
      String message =
          "Exception when trying to cleanup container " + containerIdStr
              + ": " + StringUtils.stringifyException(e);
      LOG.warn(message);
      dispatcher.getEventHandler().handle(
          new ContainerDiagnosticsUpdateEvent(containerId, message));
    } finally {
      // cleanup pid file if present
      if (pidFilePath != null) {
        try {
          FileContext lfs = FileContext.getLocalFSFileContext();
          lfs.delete(pidFilePath, false);
          lfs.delete(pidFilePath.suffix(EXIT_CODE_FILE_SUFFIX), false);
        } catch (IOException ioe) {
          LOG.warn("{} exception trying to delete pid file {}. Ignoring.",
              containerId, pidFilePath, ioe);
        }
      }
    }

    try {
      // Reap the container
      launch.reapContainer();
    } catch (IOException ioe) {
      LOG.warn("{} exception trying to reap container. Ignoring.", containerId,
          ioe);
    }
  }
@Test
  public void testFailedExitCleanup() throws Exception {
    launch.completed.set(true);
    cleanup.run();
    ArgumentCaptor<ContainerSignalContext> captor =
        ArgumentCaptor.forClass(ContainerSignalContext.class);

    verify(executor, Mockito.times(1)).signalContainer(captor.capture());
    Assert.assertEquals("signal", ContainerExecutor.Signal.TERM,
        captor.getValue().getSignal());
  }