ElasticsearchScheduler implements Scheduler { @Override public void executorLost(SchedulerDriver driver, Protos.ExecutorID executorId, Protos.SlaveID slaveId, int status) { LOGGER.info("Executor lost: " + executorId.getValue() + " on slave " + slaveId.getValue() + " with status " + status); try { Protos.TaskInfo taskInfo = clusterState.getTask(executorId); statusUpdate(driver, Protos.TaskStatus.newBuilder().setExecutorId(executorId).setSlaveId(slaveId).setTaskId(taskInfo.getTaskId()).setState(Protos.TaskState.TASK_LOST).build()); driver.killTask(taskInfo.getTaskId()); } catch (IllegalArgumentException e) { LOGGER.warn("Unable to find TaskInfo with the given Executor ID", e); } } ElasticsearchScheduler(Configuration configuration, FrameworkState frameworkState, ClusterState clusterState, TaskInfoFactory taskInfoFactory, OfferStrategy offerStrategy, SerializableState zookeeperStateDriver); Map<String, Task> getTasks(); void run(SchedulerDriver schedulerDriver); void reapTasks(); @Override void registered(SchedulerDriver driver, Protos.FrameworkID frameworkId, Protos.MasterInfo masterInfo); @Override void reregistered(SchedulerDriver driver, Protos.MasterInfo masterInfo); @Override void resourceOffers(SchedulerDriver driver, List<Protos.Offer> offers); @Override void offerRescinded(SchedulerDriver driver, Protos.OfferID offerId); @Override void statusUpdate(SchedulerDriver driver, Protos.TaskStatus status); @Override void frameworkMessage(SchedulerDriver driver, Protos.ExecutorID executorId, Protos.SlaveID slaveId, byte[] data); @Override void disconnected(SchedulerDriver driver); @Override void slaveLost(SchedulerDriver driver, Protos.SlaveID slaveId); @Override void executorLost(SchedulerDriver driver, Protos.ExecutorID executorId, Protos.SlaveID slaveId, int status); @Override void error(SchedulerDriver driver, String message); void shutdown(SchedulerDriver driver);  }
@Test public void shouldCallObserversWhenExecutorLost() { Protos.ExecutorID executorID = ProtoTestUtil.getExecutorId(); Protos.SlaveID slaveID = ProtoTestUtil.getSlaveId(); when(clusterState.getTask(executorID)).thenReturn(ProtoTestUtil.getDefaultTaskInfo()); scheduler.executorLost(driver, executorID, slaveID, 1); verify(frameworkState).announceStatusUpdate(argThat(new ArgumentMatcher<Protos.TaskStatus>() { @Override public boolean matches(Object argument) { return ((Protos.TaskStatus) argument).getState().equals(Protos.TaskState.TASK_LOST); } })); }