@Test public void testStopAtOnceFailForMsgJobAsJobIsEnable() throws SaturnJobConsoleException { String jobName = "testJob"; when(jobService.getJobStatus(TEST_NAME_SPACE_NAME, jobName)).thenReturn(JobStatus.STOPPING); List<JobServer> servers = Lists.newArrayList(); JobServer jobServer = createJobServer("job1"); servers.add(jobServer); when(jobService.getJobServers(TEST_NAME_SPACE_NAME, jobName)).thenReturn(servers); List<String> serverNameList = getJobServerNameList(servers); when(jobService.getJobServerList(TEST_NAME_SPACE_NAME, jobName)).thenReturn(serverNameList); try { restApiService.stopJobAtOnce(TEST_NAME_SPACE_NAME, jobName); } catch (SaturnJobConsoleHttpException e) { assertEquals("status code is not 400", 400, e.getStatusCode()); assertEquals("error message is not equals", "job cannot be stopped while it is enable", e.getMessage()); } }
@Override public void stopJobAtOnce(final String namespace, final String jobName) 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.STOPPED.equals(js)) { log.debug("job is already stopped"); return; } if (JobStatus.READY.equals(js) || JobStatus.RUNNING.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), "job cannot be stopped while its status is READY or RUNNING"); } List<String> jobServerList = jobService.getJobServerList(namespace, jobName); if (CollectionUtils.isEmpty(jobServerList)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } for (String executorName : jobServerList) { String path = JobNodePath.getStopOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } curatorFrameworkOp.create(path); log.info("stopAtOnce namespace:{}, jobName:{}, executorName:{}", namespace, jobName, executorName); } } }); }
RestApiServiceImpl implements RestApiService { @Override public void stopJobAtOnce(final String namespace, final String jobName) 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.STOPPED.equals(js)) { log.debug("job is already stopped"); return; } if (JobStatus.READY.equals(js) || JobStatus.RUNNING.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), "job cannot be stopped while its status is READY or RUNNING"); } List<String> jobServerList = jobService.getJobServerList(namespace, jobName); if (CollectionUtils.isEmpty(jobServerList)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } for (String executorName : jobServerList) { String path = JobNodePath.getStopOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } curatorFrameworkOp.create(path); log.info("stopAtOnce namespace:{}, jobName:{}, executorName:{}", namespace, jobName, executorName); } } }); } }
RestApiServiceImpl implements RestApiService { @Override public void stopJobAtOnce(final String namespace, final String jobName) 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.STOPPED.equals(js)) { log.debug("job is already stopped"); return; } if (JobStatus.READY.equals(js) || JobStatus.RUNNING.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), "job cannot be stopped while its status is READY or RUNNING"); } List<String> jobServerList = jobService.getJobServerList(namespace, jobName); if (CollectionUtils.isEmpty(jobServerList)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } for (String executorName : jobServerList) { String path = JobNodePath.getStopOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } curatorFrameworkOp.create(path); log.info("stopAtOnce namespace:{}, jobName:{}, executorName:{}", namespace, jobName, executorName); } } }); }  }
RestApiServiceImpl implements RestApiService { @Override public void stopJobAtOnce(final String namespace, final String jobName) 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.STOPPED.equals(js)) { log.debug("job is already stopped"); return; } if (JobStatus.READY.equals(js) || JobStatus.RUNNING.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), "job cannot be stopped while its status is READY or RUNNING"); } List<String> jobServerList = jobService.getJobServerList(namespace, jobName); if (CollectionUtils.isEmpty(jobServerList)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } for (String executorName : jobServerList) { String path = JobNodePath.getStopOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } curatorFrameworkOp.create(path); log.info("stopAtOnce namespace:{}, jobName:{}, executorName:{}", namespace, jobName, executorName); } } }); }  @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 stopJobAtOnce(final String namespace, final String jobName) 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.STOPPED.equals(js)) { log.debug("job is already stopped"); return; } if (JobStatus.READY.equals(js) || JobStatus.RUNNING.equals(js)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), "job cannot be stopped while its status is READY or RUNNING"); } List<String> jobServerList = jobService.getJobServerList(namespace, jobName); if (CollectionUtils.isEmpty(jobServerList)) { throw new SaturnJobConsoleHttpException(HttpStatus.BAD_REQUEST.value(), NO_EXECUTOR_FOUND); } for (String executorName : jobServerList) { String path = JobNodePath.getStopOneTimePath(jobName, executorName); if (curatorFrameworkOp.checkExists(path)) { curatorFrameworkOp.delete(path); } curatorFrameworkOp.create(path); log.info("stopAtOnce namespace:{}, jobName:{}, executorName:{}", namespace, jobName, executorName); } } }); }  @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);  }