@Test public void testRunAtOnceFailAsJobStatusIsNotReady() throws SaturnJobConsoleException { String jobName = "testJob"; when(jobService.getJobStatus(TEST_NAME_SPACE_NAME, jobName)).thenReturn(JobStatus.RUNNING); try { restApiService.runJobAtOnce(TEST_NAME_SPACE_NAME, jobName, null); } catch (SaturnJobConsoleHttpException e) { assertEquals("status code is not 400", 400, e.getStatusCode()); assertEquals("error message is not equals", "job's status is not {READY}", e.getMessage()); } }
@Override public void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData) throws SaturnJobConsoleException { ReuseUtils .reuse(namespace, jobName, registryCenterService, curatorRepository, new ReuseCallBackWithoutReturn() { @Override public void call(CuratorRepository.CuratorFrameworkOp curatorFrameworkOp) throws SaturnJobConsoleException { JobStatus js = jobService.getJobStatus(namespace, jobName); if (!JobStatus.READY.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, JobStatus.READY.name())); } List<JobServerStatus> jobServersStatus = jobService.getJobServersStatus(namespace, jobName); if (CollectionUtils.isEmpty(jobServersStatus)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } boolean everExecute = false; for (JobServerStatus jobServerStatus : jobServersStatus) { if (ServerStatus.ONLINE.equals(jobServerStatus.getServerStatus())) { everExecute = true; String executorName = jobServerStatus.getExecutorName(); String path = JobNodePath.getRunOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } String triggeredDataStr = gson.toJson(triggeredData); curatorFrameworkOp.create(path, triggeredDataStr); log.info("runAtOnce namespace:{}, jobName:{}, executorName:{}, triggeredData:{}", namespace, jobName, executorName, triggeredDataStr); } } if (!everExecute) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), ALL_EXECUTORS_ARE_OFFLINE); } } }); }
RestApiServiceImpl implements RestApiService { @Override public void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData) throws SaturnJobConsoleException { ReuseUtils .reuse(namespace, jobName, registryCenterService, curatorRepository, new ReuseCallBackWithoutReturn() { @Override public void call(CuratorRepository.CuratorFrameworkOp curatorFrameworkOp) throws SaturnJobConsoleException { JobStatus js = jobService.getJobStatus(namespace, jobName); if (!JobStatus.READY.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, JobStatus.READY.name())); } List<JobServerStatus> jobServersStatus = jobService.getJobServersStatus(namespace, jobName); if (CollectionUtils.isEmpty(jobServersStatus)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } boolean everExecute = false; for (JobServerStatus jobServerStatus : jobServersStatus) { if (ServerStatus.ONLINE.equals(jobServerStatus.getServerStatus())) { everExecute = true; String executorName = jobServerStatus.getExecutorName(); String path = JobNodePath.getRunOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } String triggeredDataStr = gson.toJson(triggeredData); curatorFrameworkOp.create(path, triggeredDataStr); log.info("runAtOnce namespace:{}, jobName:{}, executorName:{}, triggeredData:{}", namespace, jobName, executorName, triggeredDataStr); } } if (!everExecute) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), ALL_EXECUTORS_ARE_OFFLINE); } } }); } }
RestApiServiceImpl implements RestApiService { @Override public void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData) throws SaturnJobConsoleException { ReuseUtils .reuse(namespace, jobName, registryCenterService, curatorRepository, new ReuseCallBackWithoutReturn() { @Override public void call(CuratorRepository.CuratorFrameworkOp curatorFrameworkOp) throws SaturnJobConsoleException { JobStatus js = jobService.getJobStatus(namespace, jobName); if (!JobStatus.READY.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, JobStatus.READY.name())); } List<JobServerStatus> jobServersStatus = jobService.getJobServersStatus(namespace, jobName); if (CollectionUtils.isEmpty(jobServersStatus)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } boolean everExecute = false; for (JobServerStatus jobServerStatus : jobServersStatus) { if (ServerStatus.ONLINE.equals(jobServerStatus.getServerStatus())) { everExecute = true; String executorName = jobServerStatus.getExecutorName(); String path = JobNodePath.getRunOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } String triggeredDataStr = gson.toJson(triggeredData); curatorFrameworkOp.create(path, triggeredDataStr); log.info("runAtOnce namespace:{}, jobName:{}, executorName:{}, triggeredData:{}", namespace, jobName, executorName, triggeredDataStr); } } if (!everExecute) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), ALL_EXECUTORS_ARE_OFFLINE); } } }); }  }
RestApiServiceImpl implements RestApiService { @Override public void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData) throws SaturnJobConsoleException { ReuseUtils .reuse(namespace, jobName, registryCenterService, curatorRepository, new ReuseCallBackWithoutReturn() { @Override public void call(CuratorRepository.CuratorFrameworkOp curatorFrameworkOp) throws SaturnJobConsoleException { JobStatus js = jobService.getJobStatus(namespace, jobName); if (!JobStatus.READY.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, JobStatus.READY.name())); } List<JobServerStatus> jobServersStatus = jobService.getJobServersStatus(namespace, jobName); if (CollectionUtils.isEmpty(jobServersStatus)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } boolean everExecute = false; for (JobServerStatus jobServerStatus : jobServersStatus) { if (ServerStatus.ONLINE.equals(jobServerStatus.getServerStatus())) { everExecute = true; String executorName = jobServerStatus.getExecutorName(); String path = JobNodePath.getRunOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } String triggeredDataStr = gson.toJson(triggeredData); curatorFrameworkOp.create(path, triggeredDataStr); log.info("runAtOnce namespace:{}, jobName:{}, executorName:{}, triggeredData:{}", namespace, jobName, executorName, triggeredDataStr); } } if (!everExecute) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), ALL_EXECUTORS_ARE_OFFLINE); } } }); }  @Override void createJob(final String namespace, final JobConfig jobConfig); @Override RestApiJobInfo getRestAPIJobInfo(String namespace, final String jobName); @Override List<RestApiJobInfo> getRestApiJobInfos(final String namespace); @Transactional(rollbackFor = Exception.class) @Override void enableJob(final String namespace, final String jobName); @Transactional(rollbackFor = Exception.class) @Override void disableJob(final String namespace, final String jobName); @Override void updateJobCron(final String namespace, final String jobName, final String cron,
			final Map<String, String> customContext); @Override void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData); @Override void stopJobAtOnce(final String namespace, final String jobName); @Override void deleteJob(final String namespace, final String jobName); @Override void raiseAlarm(final String namespace, final String jobName, final String executorName,
			final Integer shardItem, final AlarmInfo alarmInfo); @Override void raiseExecutorRestartAlarm(final String namespace, final String executorName, final AlarmInfo alarmInfo); @Override void updateJob(final String namespace, final String jobName, final JobConfig jobConfig); @Override List<BatchJobResult> runDownStream(final String namespace, final String jobName,
			final Map<String, Object> triggeredData); @Override List<JobConfig> getJobConfigList(String namespace); }
RestApiServiceImpl implements RestApiService { @Override public void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData) throws SaturnJobConsoleException { ReuseUtils .reuse(namespace, jobName, registryCenterService, curatorRepository, new ReuseCallBackWithoutReturn() { @Override public void call(CuratorRepository.CuratorFrameworkOp curatorFrameworkOp) throws SaturnJobConsoleException { JobStatus js = jobService.getJobStatus(namespace, jobName); if (!JobStatus.READY.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, JobStatus.READY.name())); } List<JobServerStatus> jobServersStatus = jobService.getJobServersStatus(namespace, jobName); if (CollectionUtils.isEmpty(jobServersStatus)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } boolean everExecute = false; for (JobServerStatus jobServerStatus : jobServersStatus) { if (ServerStatus.ONLINE.equals(jobServerStatus.getServerStatus())) { everExecute = true; String executorName = jobServerStatus.getExecutorName(); String path = JobNodePath.getRunOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } String triggeredDataStr = gson.toJson(triggeredData); curatorFrameworkOp.create(path, triggeredDataStr); log.info("runAtOnce namespace:{}, jobName:{}, executorName:{}, triggeredData:{}", namespace, jobName, executorName, triggeredDataStr); } } if (!everExecute) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), ALL_EXECUTORS_ARE_OFFLINE); } } }); }  @Override void createJob(final String namespace, final JobConfig jobConfig); @Override RestApiJobInfo getRestAPIJobInfo(String namespace, final String jobName); @Override List<RestApiJobInfo> getRestApiJobInfos(final String namespace); @Transactional(rollbackFor = Exception.class) @Override void enableJob(final String namespace, final String jobName); @Transactional(rollbackFor = Exception.class) @Override void disableJob(final String namespace, final String jobName); @Override void updateJobCron(final String namespace, final String jobName, final String cron,
			final Map<String, String> customContext); @Override void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData); @Override void stopJobAtOnce(final String namespace, final String jobName); @Override void deleteJob(final String namespace, final String jobName); @Override void raiseAlarm(final String namespace, final String jobName, final String executorName,
			final Integer shardItem, final AlarmInfo alarmInfo); @Override void raiseExecutorRestartAlarm(final String namespace, final String executorName, final AlarmInfo alarmInfo); @Override void updateJob(final String namespace, final String jobName, final JobConfig jobConfig); @Override List<BatchJobResult> runDownStream(final String namespace, final String jobName,
			final Map<String, Object> triggeredData); @Override List<JobConfig> getJobConfigList(String namespace);  }