@Test public void assertAddWhenNotExistedAndTaskIsRunning() { TaskNode taskNode = TaskNode.builder().type(ExecutionType.FAILOVER).build(); when(regCenter.isExisted("/state/failover/test_job/" + taskNode.getTaskNodePath())).thenReturn(false); when(runningService.isTaskRunning(MetaInfo.from(taskNode.getTaskNodePath()))).thenReturn(true); failoverService.add(TaskContext.from(taskNode.getTaskNodeValue())); verify(regCenter).isExisted("/state/failover/test_job/" + taskNode.getTaskNodePath()); verify(runningService).isTaskRunning(MetaInfo.from(taskNode.getTaskNodePath())); verify(regCenter, times(0)).persist("/state/failover/test_job/" + taskNode.getTaskNodePath(), taskNode.getTaskNodeValue()); }
public void add(final TaskContext taskContext) { if (regCenter.getNumChildren(FailoverNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } String failoverTaskNodePath = FailoverNode.getFailoverTaskNodePath(taskContext.getMetaInfo().toString()); if (!regCenter.isExisted(failoverTaskNodePath) && !runningService.isTaskRunning(taskContext.getMetaInfo())) { regCenter.persist(failoverTaskNodePath, taskContext.getId()); } }
FailoverService { public void add(final TaskContext taskContext) { if (regCenter.getNumChildren(FailoverNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } String failoverTaskNodePath = FailoverNode.getFailoverTaskNodePath(taskContext.getMetaInfo().toString()); if (!regCenter.isExisted(failoverTaskNodePath) && !runningService.isTaskRunning(taskContext.getMetaInfo())) { regCenter.persist(failoverTaskNodePath, taskContext.getId()); } } }
FailoverService { public void add(final TaskContext taskContext) { if (regCenter.getNumChildren(FailoverNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } String failoverTaskNodePath = FailoverNode.getFailoverTaskNodePath(taskContext.getMetaInfo().toString()); if (!regCenter.isExisted(failoverTaskNodePath) && !runningService.isTaskRunning(taskContext.getMetaInfo())) { regCenter.persist(failoverTaskNodePath, taskContext.getId()); } } FailoverService(final CoordinatorRegistryCenter regCenter); }
FailoverService { public void add(final TaskContext taskContext) { if (regCenter.getNumChildren(FailoverNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } String failoverTaskNodePath = FailoverNode.getFailoverTaskNodePath(taskContext.getMetaInfo().toString()); if (!regCenter.isExisted(failoverTaskNodePath) && !runningService.isTaskRunning(taskContext.getMetaInfo())) { regCenter.persist(failoverTaskNodePath, taskContext.getId()); } } FailoverService(final CoordinatorRegistryCenter regCenter); void add(final TaskContext taskContext); Collection<JobContext> getAllEligibleJobContexts(); void remove(final Collection<MetaInfo> metaInfoList); Optional<String> getTaskId(final MetaInfo metaInfo); Map<String, Collection<FailoverTaskInfo>> getAllFailoverTasks(); }
FailoverService { public void add(final TaskContext taskContext) { if (regCenter.getNumChildren(FailoverNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } String failoverTaskNodePath = FailoverNode.getFailoverTaskNodePath(taskContext.getMetaInfo().toString()); if (!regCenter.isExisted(failoverTaskNodePath) && !runningService.isTaskRunning(taskContext.getMetaInfo())) { regCenter.persist(failoverTaskNodePath, taskContext.getId()); } } FailoverService(final CoordinatorRegistryCenter regCenter); void add(final TaskContext taskContext); Collection<JobContext> getAllEligibleJobContexts(); void remove(final Collection<MetaInfo> metaInfoList); Optional<String> getTaskId(final MetaInfo metaInfo); Map<String, Collection<FailoverTaskInfo>> getAllFailoverTasks();  }