@Override
  public void run() {
    ExecutorService executor = null;
    try {
      // Disable OOM killer and set a limit.
      // This has to be set first, so that we get notified about valid events.
      // We will be notified about events even, if they happened before
      // oom-listener started
      setCGroupParameters();

      // Start a listener process
      ProcessBuilder oomListener = new ProcessBuilder();
      oomListener.command(oomListenerPath, yarnCGroupPath);
      synchronized (this) {
        if (!stopped) {
          process = oomListener.start();
        } else {
          resetCGroupParameters();
          LOG.info("Listener stopped before starting");
          return;
        }
      }
      LOG.info(String.format("Listening on %s with %s",
          yarnCGroupPath,
          oomListenerPath));

      // We need 1 thread for the error stream and a few others
      // as a watchdog for the OOM killer
      executor = Executors.newFixedThreadPool(2);

      // Listen to any errors in the background. We do not expect this to
      // be large in size, so it will fit into a string.
      Future<String> errorListener =
          executor.submit(() -> IOUtils.toString(process.getErrorStream(), StandardCharsets.UTF_8));

      // We get Linux event increments (8 bytes) forwarded from the event stream
      // The events cannot be split, so it is safe to read them as a whole
      // There is no race condition with the cgroup
      // running out of memory. If oom is 1 at startup
      // oom_listener will send an initial notification
      InputStream events = process.getInputStream();
      byte[] event = new byte[8];
      int read;
      // This loop can be exited by terminating the process
      // with stopListening()
      while ((read = events.read(event)) == event.length) {
        // An OOM event has occurred
        resolveOOM(executor);
      }

      if (read != -1) {
        LOG.warn(String.format("Characters returned from event hander: %d",
            read));
      }

      // If the input stream is closed, we wait for exit or process terminated.
      int exitCode = process.waitFor();
      String error = errorListener.get();
      process = null;
      LOG.info(String.format("OOM listener exited %d %s", exitCode, error));
    } catch (OOMNotResolvedException ex) {
      // We could mark the node unhealthy but it shuts down the node anyways.
      // Let's just bring down the node manager all containers are frozen.
      throw new YarnRuntimeException("Could not resolve OOM", ex);
    } catch (Exception ex) {
      synchronized (this) {
        if (!stopped) {
          LOG.warn("OOM Listener exiting.", ex);
        }
      }
    } finally {
      // Make sure we do not leak the child process,
      // especially if process.waitFor() did not finish.
      if (process != null && process.isAlive()) {
        process.destroyForcibly();
      }
      if (executor != null) {
        try {
          executor.awaitTermination(6, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
          LOG.warn("Exiting without processing all OOM events.");
        }
        executor.shutdown();
      }
      resetCGroupParameters();
    }
  }
@Test(timeout = 20000, expected = YarnRuntimeException.class)
  public void testNothingToKill() throws Exception {
    conf.set(YarnConfiguration.NM_ELASTIC_MEMORY_CONTROL_OOM_LISTENER_PATH,
        script.getAbsolutePath());
    Runnable handler = mock(Runnable.class);
    try {
      FileUtils.writeStringToFile(script, "#!/bin/bash\nprintf oomevent;sleep 1000;\n",
          StandardCharsets.UTF_8, false);
      assertTrue("Could not set executable",
          script.setExecutable(true));

      CGroupsHandler cgroups = mock(CGroupsHandler.class);
      when(cgroups.getPathForCGroup(any(), any())).thenReturn("");
      when(cgroups.getCGroupParam(any(), any(), any()))
          .thenReturn("under_oom 1");

      doThrow(new YarnRuntimeException("Expected")).when(handler).run();

      CGroupElasticMemoryController controller =
          new CGroupElasticMemoryController(
              conf,
              null,
              cgroups,
              true,
              false,
              10000,
              handler
          );
      controller.run();
    } finally {
      verify(handler, times(1)).run();
      assertTrue(String.format("Could not clean up script %s",
          script.getAbsolutePath()), script.delete());
    }
  }