TaskReaper { public void run() { try { int numToKill = clusterState.getTaskList().size() - configuration.getElasticsearchNodes(); if (numToKill > 0) { LOGGER.debug("Task reaper. There are " + numToKill + " tasks to kill"); List<Protos.TaskInfo> taskList = clusterState.getTaskList(); int size = taskList.size(); numToKill = Math.min(numToKill, size); for (int x = 0; x < numToKill; x++) { int taskIndex = size - 1 - x; LOGGER.debug("Killing task index " + taskIndex); Protos.TaskInfo killTaskInfo = taskList.get(taskIndex); Protos.TaskID killTaskId = killTaskInfo.getTaskId(); LOGGER.debug("Killing task: " + killTaskId); Protos.Status status = schedulerDriver.killTask(killTaskId); LOGGER.debug("Kill request response: " + status.toString()); } } } catch (Exception ex) { LOGGER.debug("Problem with the task reaper, but carrying on like a trooper.", ex); } } TaskReaper(SchedulerDriver schedulerDriver, Configuration configuration, ClusterState clusterState); void run();  }
@Test public void shouldRemoveTask() { TaskReaper taskReaper = new TaskReaper(driver, config, state); taskReaper.run(); verify(driver, atLeastOnce()).killTask(any()); }
@Test public void shouldNotRemoveTaskWhenEqual() { when(config.getElasticsearchNodes()).thenReturn(3); TaskReaper taskReaper = new TaskReaper(driver, config, state); taskReaper.run(); verify(driver, never()).killTask(any()); }
@Test public void shouldNotRemoveTaskWhenGreater() { when(config.getElasticsearchNodes()).thenReturn(4); TaskReaper taskReaper = new TaskReaper(driver, config, state); taskReaper.run(); verify(driver, never()).killTask(any()); }
@Test public void shouldNotKillMoreThanAvailable() { when(config.getElasticsearchNodes()).thenReturn(-1); TaskReaper taskReaper = new TaskReaper(driver, config, state); taskReaper.run(); verify(driver, times(3)).killTask(any()); }