@Test public void assertAddTransientWhenJobNotExisted() { when(configService.load("test_job")).thenReturn(Optional.of(CloudJobConfigurationBuilder.createCloudJobConfiguration("test_job"))); readyService.addTransient("test_job"); verify(regCenter).persist("/state/ready/test_job", "1"); }
public void addTransient(final String jobName) { if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add transient job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } Optional<CloudJobConfigurationPOJO> cloudJobConfig = configService.load(jobName); if (!cloudJobConfig.isPresent() || CloudJobExecutionType.TRANSIENT != cloudJobConfig.get().getJobExecutionType()) { return; } String readyJobNode = ReadyNode.getReadyJobNodePath(jobName); String times = regCenter.getDirectly(readyJobNode); if (cloudJobConfig.get().isMisfire()) { regCenter.persist(readyJobNode, Integer.toString(null == times ? 1 : Integer.parseInt(times) + 1)); } else { regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1"); } }
ReadyService { public void addTransient(final String jobName) { if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add transient job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } Optional<CloudJobConfigurationPOJO> cloudJobConfig = configService.load(jobName); if (!cloudJobConfig.isPresent() || CloudJobExecutionType.TRANSIENT != cloudJobConfig.get().getJobExecutionType()) { return; } String readyJobNode = ReadyNode.getReadyJobNodePath(jobName); String times = regCenter.getDirectly(readyJobNode); if (cloudJobConfig.get().isMisfire()) { regCenter.persist(readyJobNode, Integer.toString(null == times ? 1 : Integer.parseInt(times) + 1)); } else { regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1"); } } }
ReadyService { public void addTransient(final String jobName) { if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add transient job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } Optional<CloudJobConfigurationPOJO> cloudJobConfig = configService.load(jobName); if (!cloudJobConfig.isPresent() || CloudJobExecutionType.TRANSIENT != cloudJobConfig.get().getJobExecutionType()) { return; } String readyJobNode = ReadyNode.getReadyJobNodePath(jobName); String times = regCenter.getDirectly(readyJobNode); if (cloudJobConfig.get().isMisfire()) { regCenter.persist(readyJobNode, Integer.toString(null == times ? 1 : Integer.parseInt(times) + 1)); } else { regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1"); } } ReadyService(final CoordinatorRegistryCenter regCenter); }
ReadyService { public void addTransient(final String jobName) { if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add transient job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } Optional<CloudJobConfigurationPOJO> cloudJobConfig = configService.load(jobName); if (!cloudJobConfig.isPresent() || CloudJobExecutionType.TRANSIENT != cloudJobConfig.get().getJobExecutionType()) { return; } String readyJobNode = ReadyNode.getReadyJobNodePath(jobName); String times = regCenter.getDirectly(readyJobNode); if (cloudJobConfig.get().isMisfire()) { regCenter.persist(readyJobNode, Integer.toString(null == times ? 1 : Integer.parseInt(times) + 1)); } else { regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1"); } } ReadyService(final CoordinatorRegistryCenter regCenter); void addTransient(final String jobName); void addDaemon(final String jobName); void setMisfireDisabled(final String jobName); Collection<JobContext> getAllEligibleJobContexts(final Collection<JobContext> ineligibleJobContexts); void remove(final Collection<String> jobNames); Map<String, Integer> getAllReadyTasks(); }
ReadyService { public void addTransient(final String jobName) { if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add transient job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } Optional<CloudJobConfigurationPOJO> cloudJobConfig = configService.load(jobName); if (!cloudJobConfig.isPresent() || CloudJobExecutionType.TRANSIENT != cloudJobConfig.get().getJobExecutionType()) { return; } String readyJobNode = ReadyNode.getReadyJobNodePath(jobName); String times = regCenter.getDirectly(readyJobNode); if (cloudJobConfig.get().isMisfire()) { regCenter.persist(readyJobNode, Integer.toString(null == times ? 1 : Integer.parseInt(times) + 1)); } else { regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1"); } } ReadyService(final CoordinatorRegistryCenter regCenter); void addTransient(final String jobName); void addDaemon(final String jobName); void setMisfireDisabled(final String jobName); Collection<JobContext> getAllEligibleJobContexts(final Collection<JobContext> ineligibleJobContexts); void remove(final Collection<String> jobNames); Map<String, Integer> getAllReadyTasks();  }