SchedulerEngine implements Scheduler { @Override public void statusUpdate(final SchedulerDriver schedulerDriver, final Protos.TaskStatus taskStatus) { String taskId = taskStatus.getTaskId().getValue(); TaskContext taskContext = TaskContext.from(taskId); String jobName = taskContext.getMetaInfo().getJobName(); log.trace("call statusUpdate task state is: {}, task id is: {}", taskStatus.getState(), taskId); jobEventBus.post(new JobStatusTraceEvent(jobName, taskContext.getId(), taskContext.getSlaveId(), JobStatusTraceEvent.Source.CLOUD_SCHEDULER, taskContext.getType().toString(), String.valueOf(taskContext.getMetaInfo().getShardingItems()), JobStatusTraceEvent.State.valueOf(taskStatus.getState().name()), taskStatus.getMessage())); switch (taskStatus.getState()) { case TASK_RUNNING: if (!facadeService.load(jobName).isPresent()) { schedulerDriver.killTask(Protos.TaskID.newBuilder().setValue(taskId).build()); } if ("BEGIN".equals(taskStatus.getMessage())) { facadeService.updateDaemonStatus(taskContext, false); } else if ("COMPLETE".equals(taskStatus.getMessage())) { facadeService.updateDaemonStatus(taskContext, true); statisticManager.taskRunSuccessfully(); } break; case TASK_FINISHED: facadeService.removeRunning(taskContext); unAssignTask(taskId); statisticManager.taskRunSuccessfully(); break; case TASK_KILLED: log.warn("task id is: {}, status is: {}, message is: {}, source is: {}", taskId, taskStatus.getState(), taskStatus.getMessage(), taskStatus.getSource()); facadeService.removeRunning(taskContext); facadeService.addDaemonJobToReadyQueue(jobName); unAssignTask(taskId); break; case TASK_LOST: case TASK_DROPPED: case TASK_GONE: case TASK_GONE_BY_OPERATOR: case TASK_FAILED: case TASK_ERROR: log.warn("task id is: {}, status is: {}, message is: {}, source is: {}", taskId, taskStatus.getState(), taskStatus.getMessage(), taskStatus.getSource()); facadeService.removeRunning(taskContext); facadeService.recordFailoverTask(taskContext); unAssignTask(taskId); statisticManager.taskRunFailed(); break; case TASK_UNKNOWN: case TASK_UNREACHABLE: log.error("task id is: {}, status is: {}, message is: {}, source is: {}", taskId, taskStatus.getState(), taskStatus.getMessage(), taskStatus.getSource()); statisticManager.taskRunFailed(); break; default: break; } }  @Override void registered(final SchedulerDriver schedulerDriver, final Protos.FrameworkID frameworkID, final Protos.MasterInfo masterInfo); @Override void reregistered(final SchedulerDriver schedulerDriver, final Protos.MasterInfo masterInfo); @Override void resourceOffers(final SchedulerDriver schedulerDriver, final List<Protos.Offer> offers); @Override void offerRescinded(final SchedulerDriver schedulerDriver, final Protos.OfferID offerID); @Override void statusUpdate(final SchedulerDriver schedulerDriver, final Protos.TaskStatus taskStatus); @Override void frameworkMessage(final SchedulerDriver schedulerDriver, final Protos.ExecutorID executorID, final Protos.SlaveID slaveID, final byte[] bytes); @Override void disconnected(final SchedulerDriver schedulerDriver); @Override void slaveLost(final SchedulerDriver schedulerDriver, final Protos.SlaveID slaveID); @Override void executorLost(final SchedulerDriver schedulerDriver, final Protos.ExecutorID executorID, final Protos.SlaveID slaveID, final int i); @Override void error(final SchedulerDriver schedulerDriver, final String message);  }
@Test public void assertRunningStatusUpdateForDaemonJobBegin() { TaskNode taskNode = TaskNode.builder().build(); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_RUNNING).setMessage("BEGIN").setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).updateDaemonStatus(TaskContext.from(taskNode.getTaskNodeValue()), false); }
@Test public void assertRunningStatusUpdateForDaemonJobComplete() { TaskNode taskNode = TaskNode.builder().build(); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_RUNNING).setMessage("COMPLETE").setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).updateDaemonStatus(TaskContext.from(taskNode.getTaskNodeValue()), true); }
@Test public void assertRunningStatusUpdateForOther() { TaskNode taskNode = TaskNode.builder().build(); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_RUNNING).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService, times(0)).updateDaemonStatus(TaskContext.from(taskNode.getTaskNodeValue()), ArgumentMatchers.eq(ArgumentMatchers.anyBoolean())); }
@Test public void assertFinishedStatusUpdateWithoutLaunchedTasks() { TaskNode taskNode = TaskNode.builder().build(); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_FINISHED).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskScheduler, times(0)).getTaskUnAssigner(); }
@Test public void assertFinishedStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_FINISHED).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); verify(statisticManager).taskRunSuccessfully(); }
@Test public void assertKilledStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_KILLED).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(facadeService).addDaemonJobToReadyQueue("test_job"); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); }
@Test public void assertFailedStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_FAILED).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).recordFailoverTask(TaskContext.from(taskNode.getTaskNodeValue())); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); verify(statisticManager).taskRunFailed(); }
@Test public void assertErrorStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder().setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())) .setState(Protos.TaskState.TASK_ERROR).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).recordFailoverTask(TaskContext.from(taskNode.getTaskNodeValue())); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); verify(statisticManager).taskRunFailed(); }
@Test public void assertLostStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder() .setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())).setState(Protos.TaskState.TASK_LOST).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).recordFailoverTask(TaskContext.from(taskNode.getTaskNodeValue())); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); verify(statisticManager).taskRunFailed(); }
@Test public void assertDroppedStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder() .setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())).setState(Protos.TaskState.TASK_DROPPED) .setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).recordFailoverTask(TaskContext.from(taskNode.getTaskNodeValue())); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); verify(statisticManager).taskRunFailed(); }
@Test public void assertGoneStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder() .setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())).setState(Protos.TaskState.TASK_GONE).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).recordFailoverTask(TaskContext.from(taskNode.getTaskNodeValue())); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); verify(statisticManager).taskRunFailed(); }
@Test public void assertGoneByOperatorStatusUpdate() { @SuppressWarnings("unchecked") Action2<String, String> taskUnAssigner = mock(Action2.class); when(taskScheduler.getTaskUnAssigner()).thenReturn(taskUnAssigner); TaskNode taskNode = TaskNode.builder().build(); when(facadeService.popMapping(taskNode.getTaskNodeValue())).thenReturn("localhost"); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder() .setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())).setState(Protos.TaskState.TASK_GONE_BY_OPERATOR) .setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(facadeService).recordFailoverTask(TaskContext.from(taskNode.getTaskNodeValue())); verify(facadeService).removeRunning(TaskContext.from(taskNode.getTaskNodeValue())); verify(taskUnAssigner).call(TaskContext.getIdForUnassignedSlave(taskNode.getTaskNodeValue()), "localhost"); verify(statisticManager).taskRunFailed(); }
@Test public void assertUnknownStatusUpdate() { TaskNode taskNode = TaskNode.builder().build(); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder() .setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())).setState(Protos.TaskState.TASK_UNKNOWN) .setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(statisticManager).taskRunFailed(); }
@Test public void assertUnReachedStatusUpdate() { TaskNode taskNode = TaskNode.builder().build(); schedulerEngine.statusUpdate(null, Protos.TaskStatus.newBuilder() .setTaskId(Protos.TaskID.newBuilder().setValue(taskNode.getTaskNodeValue())).setState(Protos.TaskState.TASK_UNREACHABLE) .setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0")).build()); verify(statisticManager).taskRunFailed(); }