@Test public void testGetCandidaExecutorsByExecutorPathNotExist() throws SaturnJobConsoleException { JobConfig4DB jobConfig4DB = new JobConfig4DB(); jobConfig4DB.setJobName(jobName); when(currentJobConfigService.findConfigByNamespaceAndJobName(namespace, jobName)).thenReturn(jobConfig4DB); when(registryCenterService.getCuratorFrameworkOp(namespace)).thenReturn(curatorFrameworkOp); when(curatorFrameworkOp.checkExists(eq(SaturnExecutorsNode.getExecutorsNodePath()))).thenReturn(false); assertTrue(jobService.getCandidateExecutors(namespace, jobName).isEmpty()); }
@Override public List<ExecutorProvided> getCandidateExecutors(String namespace, String jobName) throws SaturnJobConsoleException { JobConfig4DB currentJobConfig = currentJobConfigService.findConfigByNamespaceAndJobName(namespace, jobName); if (currentJobConfig == null) { throw new SaturnJobConsoleException(ERROR_CODE_NOT_EXISTED, "不能获取该作业（" + jobName + "）可选择的优先Executor，因为该作业不存在"); } List<ExecutorProvided> executorProvidedList = new ArrayList<>(); CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = registryCenterService .getCuratorFrameworkOp(namespace); String executorsNodePath = SaturnExecutorsNode.getExecutorsNodePath(); if (!curatorFrameworkOp.checkExists(executorsNodePath)) { return executorProvidedList; } List<String> executors = curatorFrameworkOp.getChildren(executorsNodePath); if (executors == null) { executors = new ArrayList<>(); } if (!executors.isEmpty()) { for (String executor : executors) { if (curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorTaskNodePath(executor))) { continue; } ExecutorProvided executorProvided = new ExecutorProvided(); executorProvided.setType(ExecutorProvidedType.PHYSICAL); executorProvided.setExecutorName(executor); executorProvided.setNoTraffic( curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorNoTrafficNodePath(executor))); String ip = curatorFrameworkOp.getData(SaturnExecutorsNode.getExecutorIpNodePath(executor)); if (StringUtils.isNotBlank(ip)) { executorProvided.setStatus(ExecutorProvidedStatus.ONLINE); executorProvided.setIp(ip); } else { executorProvided.setStatus(ExecutorProvidedStatus.OFFLINE); } executorProvidedList.add(executorProvided); } } List<ExecutorProvided> dockerExecutorProvided = getContainerTaskIds(curatorFrameworkOp); executorProvidedList.addAll(dockerExecutorProvided); if (StringUtils.isBlank(jobName)) { return executorProvidedList; } String preferListNodePath = JobNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PREFER_LIST); if (!curatorFrameworkOp.checkExists(preferListNodePath)) { return executorProvidedList; } String preferList = curatorFrameworkOp.getData(preferListNodePath); if (Strings.isNullOrEmpty(preferList)) { return executorProvidedList; } handlerPreferListString(curatorFrameworkOp, preferList, executors, dockerExecutorProvided, executorProvidedList); return executorProvidedList; }
JobServiceImpl implements JobService { @Override public List<ExecutorProvided> getCandidateExecutors(String namespace, String jobName) throws SaturnJobConsoleException { JobConfig4DB currentJobConfig = currentJobConfigService.findConfigByNamespaceAndJobName(namespace, jobName); if (currentJobConfig == null) { throw new SaturnJobConsoleException(ERROR_CODE_NOT_EXISTED, "不能获取该作业（" + jobName + "）可选择的优先Executor，因为该作业不存在"); } List<ExecutorProvided> executorProvidedList = new ArrayList<>(); CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = registryCenterService .getCuratorFrameworkOp(namespace); String executorsNodePath = SaturnExecutorsNode.getExecutorsNodePath(); if (!curatorFrameworkOp.checkExists(executorsNodePath)) { return executorProvidedList; } List<String> executors = curatorFrameworkOp.getChildren(executorsNodePath); if (executors == null) { executors = new ArrayList<>(); } if (!executors.isEmpty()) { for (String executor : executors) { if (curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorTaskNodePath(executor))) { continue; } ExecutorProvided executorProvided = new ExecutorProvided(); executorProvided.setType(ExecutorProvidedType.PHYSICAL); executorProvided.setExecutorName(executor); executorProvided.setNoTraffic( curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorNoTrafficNodePath(executor))); String ip = curatorFrameworkOp.getData(SaturnExecutorsNode.getExecutorIpNodePath(executor)); if (StringUtils.isNotBlank(ip)) { executorProvided.setStatus(ExecutorProvidedStatus.ONLINE); executorProvided.setIp(ip); } else { executorProvided.setStatus(ExecutorProvidedStatus.OFFLINE); } executorProvidedList.add(executorProvided); } } List<ExecutorProvided> dockerExecutorProvided = getContainerTaskIds(curatorFrameworkOp); executorProvidedList.addAll(dockerExecutorProvided); if (StringUtils.isBlank(jobName)) { return executorProvidedList; } String preferListNodePath = JobNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PREFER_LIST); if (!curatorFrameworkOp.checkExists(preferListNodePath)) { return executorProvidedList; } String preferList = curatorFrameworkOp.getData(preferListNodePath); if (Strings.isNullOrEmpty(preferList)) { return executorProvidedList; } handlerPreferListString(curatorFrameworkOp, preferList, executors, dockerExecutorProvided, executorProvidedList); return executorProvidedList; } }
JobServiceImpl implements JobService { @Override public List<ExecutorProvided> getCandidateExecutors(String namespace, String jobName) throws SaturnJobConsoleException { JobConfig4DB currentJobConfig = currentJobConfigService.findConfigByNamespaceAndJobName(namespace, jobName); if (currentJobConfig == null) { throw new SaturnJobConsoleException(ERROR_CODE_NOT_EXISTED, "不能获取该作业（" + jobName + "）可选择的优先Executor，因为该作业不存在"); } List<ExecutorProvided> executorProvidedList = new ArrayList<>(); CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = registryCenterService .getCuratorFrameworkOp(namespace); String executorsNodePath = SaturnExecutorsNode.getExecutorsNodePath(); if (!curatorFrameworkOp.checkExists(executorsNodePath)) { return executorProvidedList; } List<String> executors = curatorFrameworkOp.getChildren(executorsNodePath); if (executors == null) { executors = new ArrayList<>(); } if (!executors.isEmpty()) { for (String executor : executors) { if (curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorTaskNodePath(executor))) { continue; } ExecutorProvided executorProvided = new ExecutorProvided(); executorProvided.setType(ExecutorProvidedType.PHYSICAL); executorProvided.setExecutorName(executor); executorProvided.setNoTraffic( curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorNoTrafficNodePath(executor))); String ip = curatorFrameworkOp.getData(SaturnExecutorsNode.getExecutorIpNodePath(executor)); if (StringUtils.isNotBlank(ip)) { executorProvided.setStatus(ExecutorProvidedStatus.ONLINE); executorProvided.setIp(ip); } else { executorProvided.setStatus(ExecutorProvidedStatus.OFFLINE); } executorProvidedList.add(executorProvided); } } List<ExecutorProvided> dockerExecutorProvided = getContainerTaskIds(curatorFrameworkOp); executorProvidedList.addAll(dockerExecutorProvided); if (StringUtils.isBlank(jobName)) { return executorProvidedList; } String preferListNodePath = JobNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PREFER_LIST); if (!curatorFrameworkOp.checkExists(preferListNodePath)) { return executorProvidedList; } String preferList = curatorFrameworkOp.getData(preferListNodePath); if (Strings.isNullOrEmpty(preferList)) { return executorProvidedList; } handlerPreferListString(curatorFrameworkOp, preferList, executors, dockerExecutorProvided, executorProvidedList); return executorProvidedList; }  }
JobServiceImpl implements JobService { @Override public List<ExecutorProvided> getCandidateExecutors(String namespace, String jobName) throws SaturnJobConsoleException { JobConfig4DB currentJobConfig = currentJobConfigService.findConfigByNamespaceAndJobName(namespace, jobName); if (currentJobConfig == null) { throw new SaturnJobConsoleException(ERROR_CODE_NOT_EXISTED, "不能获取该作业（" + jobName + "）可选择的优先Executor，因为该作业不存在"); } List<ExecutorProvided> executorProvidedList = new ArrayList<>(); CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = registryCenterService .getCuratorFrameworkOp(namespace); String executorsNodePath = SaturnExecutorsNode.getExecutorsNodePath(); if (!curatorFrameworkOp.checkExists(executorsNodePath)) { return executorProvidedList; } List<String> executors = curatorFrameworkOp.getChildren(executorsNodePath); if (executors == null) { executors = new ArrayList<>(); } if (!executors.isEmpty()) { for (String executor : executors) { if (curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorTaskNodePath(executor))) { continue; } ExecutorProvided executorProvided = new ExecutorProvided(); executorProvided.setType(ExecutorProvidedType.PHYSICAL); executorProvided.setExecutorName(executor); executorProvided.setNoTraffic( curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorNoTrafficNodePath(executor))); String ip = curatorFrameworkOp.getData(SaturnExecutorsNode.getExecutorIpNodePath(executor)); if (StringUtils.isNotBlank(ip)) { executorProvided.setStatus(ExecutorProvidedStatus.ONLINE); executorProvided.setIp(ip); } else { executorProvided.setStatus(ExecutorProvidedStatus.OFFLINE); } executorProvidedList.add(executorProvided); } } List<ExecutorProvided> dockerExecutorProvided = getContainerTaskIds(curatorFrameworkOp); executorProvidedList.addAll(dockerExecutorProvided); if (StringUtils.isBlank(jobName)) { return executorProvidedList; } String preferListNodePath = JobNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PREFER_LIST); if (!curatorFrameworkOp.checkExists(preferListNodePath)) { return executorProvidedList; } String preferList = curatorFrameworkOp.getData(preferListNodePath); if (Strings.isNullOrEmpty(preferList)) { return executorProvidedList; } handlerPreferListString(curatorFrameworkOp, preferList, executors, dockerExecutorProvided, executorProvidedList); return executorProvidedList; }  @Override List<String> getGroups(String namespace); @Transactional(rollbackFor = Exception.class) @Override void enableJob(String namespace, String jobName, String updatedBy); @Transactional(rollbackFor = Exception.class) @Override void disableJob(String namespace, String jobName, String updatedBy); @Transactional(rollbackFor = Exception.class) @Override void removeJob(String namespace, String jobName); @Override List<ExecutorProvided> getCandidateExecutors(String namespace, String jobName); @Transactional(rollbackFor = Exception.class) @Override void setPreferList(String namespace, String jobName, String preferList, String updatedBy); @Override List<String> getCandidateUpStream(String namespace); @Override List<String> getCandidateDownStream(String namespace); @Transactional(rollbackFor = Exception.class) @Override void addJob(String namespace, JobConfig jobConfig, String createdBy); @Transactional(rollbackFor = Exception.class) @Override void copyJob(String namespace, JobConfig jobConfig, String jobNameCopied, String createdBy); @Override int getMaxJobNum(); @Override boolean jobIncExceeds(String namespace, int maxJobNum, int inc); @Override List<JobConfig> getUnSystemJobs(String namespace); @Override List<JobConfig4DB> getUnSystemJobsWithCondition(String namespace, Map<String, Object> condition, int page,
			int size); @Override int countUnSystemJobsWithCondition(String namespace, Map<String, Object> condition); @Override int countEnabledUnSystemJobs(String namespace); @Override List<String> getUnSystemJobNames(String namespace); @Override List<String> getJobNames(String namespace); @Override void persistJobFromDB(String namespace, JobConfig jobConfig); @Override void persistJobFromDB(JobConfig jobConfig, CuratorFrameworkOp curatorFrameworkOp); @Override List<BatchJobResult> importJobs(String namespace, MultipartFile file, String createdBy); @Override File exportJobs(String namespace); @Override File exportSelectedJobs(String namespace, List<String> jobList); @Override ArrangeLayout getArrangeLayout(String namespace); @Override JobConfig getJobConfigFromZK(String namespace, String jobName); @Override JobConfig getJobConfig(String namespace, String jobName); @Override JobStatus getJobStatus(String namespace, String jobName); @Override JobStatus getJobStatus(String namespace, JobConfig jobConfig); @Override boolean isJobShardingAllocatedExecutor(String namespace, String jobName); @Override List<String> getJobServerList(String namespace, String jobName); @Override GetJobConfigVo getJobConfigVo(String namespace, String jobName); @Transactional(rollbackFor = Exception.class) @Override void updateJobConfig(String namespace, JobConfig jobConfig, String updatedBy); @Override List<String> getAllJobNamesFromZK(String namespace); @Transactional(rollbackFor = Exception.class) @Override void updateJobCron(String namespace, String jobName, String cron, Map<String, String> customContext,
			String updatedBy); @Override List<JobServer> getJobServers(String namespace, String jobName); @Override List<JobServerStatus> getJobServersStatus(String namespace, String jobName); @Override void runAtOnce(String namespace, String jobName); @Override void stopAtOnce(String namespace, String jobName); @Override List<ExecutionInfo> getExecutionStatus(String namespace, String jobName); @Override String getExecutionLog(String namespace, String jobName, String jobItem); @Override List<JobConfig4DB> getJobsByQueue(String queue); @Transactional(rollbackFor = Exception.class) @Override void batchSetGroups(String namespace, List<String> jobNames, List<String> oldGroupNames,
			List<String> newGroupNames, String userName); }
JobServiceImpl implements JobService { @Override public List<ExecutorProvided> getCandidateExecutors(String namespace, String jobName) throws SaturnJobConsoleException { JobConfig4DB currentJobConfig = currentJobConfigService.findConfigByNamespaceAndJobName(namespace, jobName); if (currentJobConfig == null) { throw new SaturnJobConsoleException(ERROR_CODE_NOT_EXISTED, "不能获取该作业（" + jobName + "）可选择的优先Executor，因为该作业不存在"); } List<ExecutorProvided> executorProvidedList = new ArrayList<>(); CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = registryCenterService .getCuratorFrameworkOp(namespace); String executorsNodePath = SaturnExecutorsNode.getExecutorsNodePath(); if (!curatorFrameworkOp.checkExists(executorsNodePath)) { return executorProvidedList; } List<String> executors = curatorFrameworkOp.getChildren(executorsNodePath); if (executors == null) { executors = new ArrayList<>(); } if (!executors.isEmpty()) { for (String executor : executors) { if (curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorTaskNodePath(executor))) { continue; } ExecutorProvided executorProvided = new ExecutorProvided(); executorProvided.setType(ExecutorProvidedType.PHYSICAL); executorProvided.setExecutorName(executor); executorProvided.setNoTraffic( curatorFrameworkOp.checkExists(SaturnExecutorsNode.getExecutorNoTrafficNodePath(executor))); String ip = curatorFrameworkOp.getData(SaturnExecutorsNode.getExecutorIpNodePath(executor)); if (StringUtils.isNotBlank(ip)) { executorProvided.setStatus(ExecutorProvidedStatus.ONLINE); executorProvided.setIp(ip); } else { executorProvided.setStatus(ExecutorProvidedStatus.OFFLINE); } executorProvidedList.add(executorProvided); } } List<ExecutorProvided> dockerExecutorProvided = getContainerTaskIds(curatorFrameworkOp); executorProvidedList.addAll(dockerExecutorProvided); if (StringUtils.isBlank(jobName)) { return executorProvidedList; } String preferListNodePath = JobNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PREFER_LIST); if (!curatorFrameworkOp.checkExists(preferListNodePath)) { return executorProvidedList; } String preferList = curatorFrameworkOp.getData(preferListNodePath); if (Strings.isNullOrEmpty(preferList)) { return executorProvidedList; } handlerPreferListString(curatorFrameworkOp, preferList, executors, dockerExecutorProvided, executorProvidedList); return executorProvidedList; }  @Override List<String> getGroups(String namespace); @Transactional(rollbackFor = Exception.class) @Override void enableJob(String namespace, String jobName, String updatedBy); @Transactional(rollbackFor = Exception.class) @Override void disableJob(String namespace, String jobName, String updatedBy); @Transactional(rollbackFor = Exception.class) @Override void removeJob(String namespace, String jobName); @Override List<ExecutorProvided> getCandidateExecutors(String namespace, String jobName); @Transactional(rollbackFor = Exception.class) @Override void setPreferList(String namespace, String jobName, String preferList, String updatedBy); @Override List<String> getCandidateUpStream(String namespace); @Override List<String> getCandidateDownStream(String namespace); @Transactional(rollbackFor = Exception.class) @Override void addJob(String namespace, JobConfig jobConfig, String createdBy); @Transactional(rollbackFor = Exception.class) @Override void copyJob(String namespace, JobConfig jobConfig, String jobNameCopied, String createdBy); @Override int getMaxJobNum(); @Override boolean jobIncExceeds(String namespace, int maxJobNum, int inc); @Override List<JobConfig> getUnSystemJobs(String namespace); @Override List<JobConfig4DB> getUnSystemJobsWithCondition(String namespace, Map<String, Object> condition, int page,
			int size); @Override int countUnSystemJobsWithCondition(String namespace, Map<String, Object> condition); @Override int countEnabledUnSystemJobs(String namespace); @Override List<String> getUnSystemJobNames(String namespace); @Override List<String> getJobNames(String namespace); @Override void persistJobFromDB(String namespace, JobConfig jobConfig); @Override void persistJobFromDB(JobConfig jobConfig, CuratorFrameworkOp curatorFrameworkOp); @Override List<BatchJobResult> importJobs(String namespace, MultipartFile file, String createdBy); @Override File exportJobs(String namespace); @Override File exportSelectedJobs(String namespace, List<String> jobList); @Override ArrangeLayout getArrangeLayout(String namespace); @Override JobConfig getJobConfigFromZK(String namespace, String jobName); @Override JobConfig getJobConfig(String namespace, String jobName); @Override JobStatus getJobStatus(String namespace, String jobName); @Override JobStatus getJobStatus(String namespace, JobConfig jobConfig); @Override boolean isJobShardingAllocatedExecutor(String namespace, String jobName); @Override List<String> getJobServerList(String namespace, String jobName); @Override GetJobConfigVo getJobConfigVo(String namespace, String jobName); @Transactional(rollbackFor = Exception.class) @Override void updateJobConfig(String namespace, JobConfig jobConfig, String updatedBy); @Override List<String> getAllJobNamesFromZK(String namespace); @Transactional(rollbackFor = Exception.class) @Override void updateJobCron(String namespace, String jobName, String cron, Map<String, String> customContext,
			String updatedBy); @Override List<JobServer> getJobServers(String namespace, String jobName); @Override List<JobServerStatus> getJobServersStatus(String namespace, String jobName); @Override void runAtOnce(String namespace, String jobName); @Override void stopAtOnce(String namespace, String jobName); @Override List<ExecutionInfo> getExecutionStatus(String namespace, String jobName); @Override String getExecutionLog(String namespace, String jobName, String jobItem); @Override List<JobConfig4DB> getJobsByQueue(String queue); @Transactional(rollbackFor = Exception.class) @Override void batchSetGroups(String namespace, List<String> jobNames, List<String> oldGroupNames,
			List<String> newGroupNames, String userName); static final String CONFIG_ITEM_LOAD_LEVEL; static final String CONFIG_ITEM_ENABLED; static final String CONFIG_ITEM_DESCRIPTION; static final String CONFIG_ITEM_CUSTOM_CONTEXT; static final String CONFIG_ITEM_JOB_TYPE; static final String CONFIG_ITEM_JOB_MODE; static final String CONFIG_ITEM_SHARDING_ITEM_PARAMETERS; static final String CONFIG_ITEM_JOB_PARAMETER; static final String CONFIG_ITEM_QUEUE_NAME; static final String CONFIG_ITEM_CHANNEL_NAME; static final String CONFIG_ITEM_FAILOVER; static final String CONFIG_ITEM_MONITOR_EXECUTION; static final String CONFIG_ITEM_TIMEOUT_4_ALARM_SECONDS; static final String CONFIG_ITEM_TIMEOUT_SECONDS; static final String CONFIG_ITEM_TIME_ZONE; static final String CONFIG_ITEM_CRON; static final String CONFIG_ITEM_PAUSE_PERIOD_DATE; static final String CONFIG_ITEM_PAUSE_PERIOD_TIME; static final String CONFIG_ITEM_PROCESS_COUNT_INTERVAL_SECONDS; static final String CONFIG_ITEM_SHARDING_TOTAL_COUNT; static final String CONFIG_ITEM_SHOW_NORMAL_LOG; static final String CONFIG_ITEM_JOB_DEGREE; static final String CONFIG_ITEM_ENABLED_REPORT; static final String CONFIG_ITEM_PREFER_LIST; static final String CONFIG_ITEM_USE_DISPREFER_LIST; static final String CONFIG_ITEM_LOCAL_MODE; static final String CONFIG_ITEM_USE_SERIAL; static final String CONFIG_ITEM_DEPENDENCIES; static final String CONFIG_ITEM_GROUPS; static final String CONFIG_ITEM_JOB_CLASS; static final String CONFIG_ITEM_RERUN; static final String CONFIG_ITEM_DOWNSTREAM; static final String CONFIG_ITEM_UPSTREAM; }