@Test public void getExecutorRunningInfo() throws SaturnJobConsoleException { String namespace = "www.abc.com"; String executorName = "exec01"; when(registryCenterService.getCuratorFrameworkOp(namespace)).thenReturn(curatorFrameworkOp); JobConfig jobConfigA = new JobConfig(); jobConfigA.setJobName("jobA"); jobConfigA.setJobType(JobType.JAVA_JOB.name()); jobConfigA.setEnabledReport(Boolean.TRUE); jobConfigA.setFailover(Boolean.TRUE); jobConfigA.setLocalMode(Boolean.FALSE); JobConfig jobConfigB = new JobConfig(); jobConfigB.setJobName("jobB"); jobConfigB.setJobType(JobType.MSG_JOB.name()); jobConfigB.setEnabledReport(Boolean.FALSE); jobConfigB.setFailover(Boolean.FALSE); jobConfigB.setLocalMode(Boolean.FALSE); JobConfig jobConfigC = new JobConfig(); jobConfigC.setJobName("jobC"); jobConfigC.setJobType(JobType.JAVA_JOB.name()); jobConfigC.setEnabledReport(Boolean.TRUE); jobConfigC.setFailover(Boolean.FALSE); jobConfigC.setLocalMode(Boolean.FALSE); when(jobService.getUnSystemJobs(namespace)).thenReturn(Lists.newArrayList(jobConfigA, jobConfigB, jobConfigC)); when(curatorFrameworkOp.checkExists(String.format("/%s/%s/servers", $JOBS_NODE_NAME, jobConfigA.getJobName()))) .thenReturn(true); when(curatorFrameworkOp.checkExists(String.format("/%s/%s/servers", $JOBS_NODE_NAME, jobConfigB.getJobName()))) .thenReturn(true); when(curatorFrameworkOp.checkExists(String.format("/%s/%s/servers", $JOBS_NODE_NAME, jobConfigC.getJobName()))) .thenReturn(true); when(curatorFrameworkOp.getData( String.format("%s/%s/%s", JobNodePath.getServerNodePath(jobConfigA.getJobName()), executorName, "sharding"))).thenReturn("1,2"); when(curatorFrameworkOp.getData( String.format("%s/%s/%s", JobNodePath.getServerNodePath(jobConfigB.getJobName()), executorName, "sharding"))).thenReturn("2,3"); when(curatorFrameworkOp.getData( String.format("%s/%s/%s", JobNodePath.getServerNodePath(jobConfigC.getJobName()), executorName, "sharding"))).thenReturn("0,1"); when(curatorFrameworkOp.getChildren(JobNodePath.getExecutionNodePath("jobA"))) .thenReturn(Lists.newArrayList("0", "1", "2", "3")); when(curatorFrameworkOp.getChildren(JobNodePath.getExecutionNodePath("jobB"))) .thenReturn(Lists.newArrayList("0", "1", "2", "3")); when(curatorFrameworkOp.getChildren(JobNodePath.getExecutionNodePath("jobC"))) .thenReturn(Lists.newArrayList("0", "1", "2", "3")); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobA", "0", "running"))).thenReturn(true); when(curatorFrameworkOp.getData(JobNodePath.getExecutionNodePath("jobA", "0", "failover"))) .thenReturn(executorName); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobA", "1", "running"))).thenReturn(true); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobA", "2", "running"))).thenReturn(true); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobA", "3", "running"))).thenReturn(true); when(curatorFrameworkOp.getData(JobNodePath.getExecutionNodePath("jobA", "3", "failover"))) .thenReturn("other-exec"); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobC", "0", "running"))) .thenReturn(false); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobC", "1", "running"))).thenReturn(true); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobC", "2", "running"))).thenReturn(true); when(curatorFrameworkOp.checkExists(JobNodePath.getExecutionNodePath("jobC", "3", "running"))).thenReturn(true); ServerRunningInfo serverRunningInfo = executorService.getExecutorRunningInfo(namespace, executorName); assertEquals(2, serverRunningInfo.getRunningJobItems().size()); assertEquals("0,1,2", serverRunningInfo.getRunningJobItems().get("jobA")); assertEquals("1", serverRunningInfo.getRunningJobItems().get("jobC")); assertEquals(0, serverRunningInfo.getPotentialRunningJobItems().size()); assertEquals(null, serverRunningInfo.getPotentialRunningJobItems().get("jobB")); }
@Override public ServerRunningInfo getExecutorRunningInfo(String namespace, String executorName) throws SaturnJobConsoleException { CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = getCuratorFrameworkOp(namespace); List<JobConfig> unSystemJobs = jobService.getUnSystemJobs(namespace); ServerRunningInfo serverRunningInfo = new ServerRunningInfo(executorName); for (JobConfig jobConfig : unSystemJobs) { boolean needToCheckFailover = needToCheckFailover(jobConfig); String jobName = jobConfig.getJobName(); String serverNodePath = JobNodePath.getServerNodePath(jobName); if (!curatorFrameworkOp.checkExists(serverNodePath)) { continue; } String sharding = curatorFrameworkOp .getData(JobNodePath.getServerNodePath(jobName, executorName, "sharding")); Set<String> shardingItems = getShardingItems(sharding); if (needToCheckFailover) { obtainServerRunningInfoWhileNeedToCheckFailover(executorName, curatorFrameworkOp, jobName, shardingItems, serverRunningInfo); } else if (!shardingItems.isEmpty()) { obtainServerRunningInfoWhileNoNeedToCheckFailover(curatorFrameworkOp, jobConfig, shardingItems, serverRunningInfo); } } return serverRunningInfo; }
ExecutorServiceImpl implements ExecutorService { @Override public ServerRunningInfo getExecutorRunningInfo(String namespace, String executorName) throws SaturnJobConsoleException { CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = getCuratorFrameworkOp(namespace); List<JobConfig> unSystemJobs = jobService.getUnSystemJobs(namespace); ServerRunningInfo serverRunningInfo = new ServerRunningInfo(executorName); for (JobConfig jobConfig : unSystemJobs) { boolean needToCheckFailover = needToCheckFailover(jobConfig); String jobName = jobConfig.getJobName(); String serverNodePath = JobNodePath.getServerNodePath(jobName); if (!curatorFrameworkOp.checkExists(serverNodePath)) { continue; } String sharding = curatorFrameworkOp .getData(JobNodePath.getServerNodePath(jobName, executorName, "sharding")); Set<String> shardingItems = getShardingItems(sharding); if (needToCheckFailover) { obtainServerRunningInfoWhileNeedToCheckFailover(executorName, curatorFrameworkOp, jobName, shardingItems, serverRunningInfo); } else if (!shardingItems.isEmpty()) { obtainServerRunningInfoWhileNoNeedToCheckFailover(curatorFrameworkOp, jobConfig, shardingItems, serverRunningInfo); } } return serverRunningInfo; } }
ExecutorServiceImpl implements ExecutorService { @Override public ServerRunningInfo getExecutorRunningInfo(String namespace, String executorName) throws SaturnJobConsoleException { CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = getCuratorFrameworkOp(namespace); List<JobConfig> unSystemJobs = jobService.getUnSystemJobs(namespace); ServerRunningInfo serverRunningInfo = new ServerRunningInfo(executorName); for (JobConfig jobConfig : unSystemJobs) { boolean needToCheckFailover = needToCheckFailover(jobConfig); String jobName = jobConfig.getJobName(); String serverNodePath = JobNodePath.getServerNodePath(jobName); if (!curatorFrameworkOp.checkExists(serverNodePath)) { continue; } String sharding = curatorFrameworkOp .getData(JobNodePath.getServerNodePath(jobName, executorName, "sharding")); Set<String> shardingItems = getShardingItems(sharding); if (needToCheckFailover) { obtainServerRunningInfoWhileNeedToCheckFailover(executorName, curatorFrameworkOp, jobName, shardingItems, serverRunningInfo); } else if (!shardingItems.isEmpty()) { obtainServerRunningInfoWhileNoNeedToCheckFailover(curatorFrameworkOp, jobConfig, shardingItems, serverRunningInfo); } } return serverRunningInfo; }  }
ExecutorServiceImpl implements ExecutorService { @Override public ServerRunningInfo getExecutorRunningInfo(String namespace, String executorName) throws SaturnJobConsoleException { CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = getCuratorFrameworkOp(namespace); List<JobConfig> unSystemJobs = jobService.getUnSystemJobs(namespace); ServerRunningInfo serverRunningInfo = new ServerRunningInfo(executorName); for (JobConfig jobConfig : unSystemJobs) { boolean needToCheckFailover = needToCheckFailover(jobConfig); String jobName = jobConfig.getJobName(); String serverNodePath = JobNodePath.getServerNodePath(jobName); if (!curatorFrameworkOp.checkExists(serverNodePath)) { continue; } String sharding = curatorFrameworkOp .getData(JobNodePath.getServerNodePath(jobName, executorName, "sharding")); Set<String> shardingItems = getShardingItems(sharding); if (needToCheckFailover) { obtainServerRunningInfoWhileNeedToCheckFailover(executorName, curatorFrameworkOp, jobName, shardingItems, serverRunningInfo); } else if (!shardingItems.isEmpty()) { obtainServerRunningInfoWhileNoNeedToCheckFailover(curatorFrameworkOp, jobConfig, shardingItems, serverRunningInfo); } } return serverRunningInfo; }  @Override List<ServerBriefInfo> getExecutors(String namespace); @Override List<ServerBriefInfo> getExecutors(String namespace, ServerStatus expectedServerStatus); @Override ServerBriefInfo getExecutor(String namespace, String executorName); @Override ServerAllocationInfo getExecutorAllocation(String namespace, String executorName); @Override ServerRunningInfo getExecutorRunningInfo(String namespace, String executorName); @Override void extractTraffic(String namespace, String executorName); @Override void recoverTraffic(String namespace, String executorName); @Override void removeExecutor(String namespace, String executorName); @Override void shardAll(String namespace); @Override void dump(String namespace, String executorName); @Override File dumpAsFile(String namespace, String executorName); @Override void restart(String namespace, String executorName); }
ExecutorServiceImpl implements ExecutorService { @Override public ServerRunningInfo getExecutorRunningInfo(String namespace, String executorName) throws SaturnJobConsoleException { CuratorRepository.CuratorFrameworkOp curatorFrameworkOp = getCuratorFrameworkOp(namespace); List<JobConfig> unSystemJobs = jobService.getUnSystemJobs(namespace); ServerRunningInfo serverRunningInfo = new ServerRunningInfo(executorName); for (JobConfig jobConfig : unSystemJobs) { boolean needToCheckFailover = needToCheckFailover(jobConfig); String jobName = jobConfig.getJobName(); String serverNodePath = JobNodePath.getServerNodePath(jobName); if (!curatorFrameworkOp.checkExists(serverNodePath)) { continue; } String sharding = curatorFrameworkOp .getData(JobNodePath.getServerNodePath(jobName, executorName, "sharding")); Set<String> shardingItems = getShardingItems(sharding); if (needToCheckFailover) { obtainServerRunningInfoWhileNeedToCheckFailover(executorName, curatorFrameworkOp, jobName, shardingItems, serverRunningInfo); } else if (!shardingItems.isEmpty()) { obtainServerRunningInfoWhileNoNeedToCheckFailover(curatorFrameworkOp, jobConfig, shardingItems, serverRunningInfo); } } return serverRunningInfo; }  @Override List<ServerBriefInfo> getExecutors(String namespace); @Override List<ServerBriefInfo> getExecutors(String namespace, ServerStatus expectedServerStatus); @Override ServerBriefInfo getExecutor(String namespace, String executorName); @Override ServerAllocationInfo getExecutorAllocation(String namespace, String executorName); @Override ServerRunningInfo getExecutorRunningInfo(String namespace, String executorName); @Override void extractTraffic(String namespace, String executorName); @Override void recoverTraffic(String namespace, String executorName); @Override void removeExecutor(String namespace, String executorName); @Override void shardAll(String namespace); @Override void dump(String namespace, String executorName); @Override File dumpAsFile(String namespace, String executorName); @Override void restart(String namespace, String executorName);  }