@Test public void testJobSchedulerInit() throws Exception { File specDir = Files.createTempDir(); Properties properties = new Properties(); properties.setProperty(FLOWSPEC_STORE_DIR_KEY, specDir.getAbsolutePath()); FlowCatalog flowCatalog = new FlowCatalog(ConfigUtils.propertiesToConfig(properties)); ServiceBasedAppLauncher serviceLauncher = new ServiceBasedAppLauncher(properties, "GaaSJobSchedulerTest"); serviceLauncher.addService(flowCatalog); serviceLauncher.start(); FlowSpec flowSpec0 = FlowCatalogTest.initFlowSpec(specDir.getAbsolutePath(), URI.create("spec0")); FlowSpec flowSpec1 = FlowCatalogTest.initFlowSpec(specDir.getAbsolutePath(), URI.create("spec1")); flowCatalog.put(flowSpec0, false); flowCatalog.put(flowSpec1, false); Assert.assertEquals(flowCatalog.getSpecs().size(), 2); Orchestrator mockOrchestrator = Mockito.mock(Orchestrator.class); TestGobblinServiceJobScheduler scheduler = new TestGobblinServiceJobScheduler("testscheduler", ConfigFactory.empty(), Optional.of(flowCatalog), null, mockOrchestrator, null); SpecCompiler mockCompiler = Mockito.mock(SpecCompiler.class); Mockito.when(mockOrchestrator.getSpecCompiler()).thenReturn(mockCompiler); Mockito.doAnswer((Answer<Void>) a -> { scheduler.isCompilerHealthy = true; return null; }).when(mockCompiler).awaitHealthy(); scheduler.setActive(true); AssertWithBackoff.create().timeoutMs(6000).maxSleepMs(2000).backoffFactor(2) .assertTrue(new Predicate<Void>() { @Override public boolean apply(Void input) { Map<String, Spec> scheduledFlowSpecs = scheduler.scheduledFlowSpecs; if (scheduledFlowSpecs != null && scheduledFlowSpecs.size() == 2) { return scheduler.scheduledFlowSpecs.containsKey("spec0") && scheduler.scheduledFlowSpecs.containsKey("spec1"); } else { return false; } } }, "Waiting all flowSpecs to be scheduled"); }
public synchronized void setActive(boolean isActive) { if (this.isActive == isActive) { return; } if (isActive) { this.isActive = isActive; if (this.flowCatalog.isPresent()) { Thread scheduleSpec = new Thread(new Runnable() { @Override public void run() { try { GobblinServiceJobScheduler.this.orchestrator.getSpecCompiler().awaitHealthy(); } catch (InterruptedException e) { throw new RuntimeException(e); } scheduleSpecsFromCatalog(); } }); scheduleSpec.start(); } } else { for (Spec spec : this.scheduledFlowSpecs.values()) { onDeleteSpec(spec.getUri(), spec.getVersion()); } this.isActive = isActive; } }
GobblinServiceJobScheduler extends JobScheduler implements SpecCatalogListener { public synchronized void setActive(boolean isActive) { if (this.isActive == isActive) { return; } if (isActive) { this.isActive = isActive; if (this.flowCatalog.isPresent()) { Thread scheduleSpec = new Thread(new Runnable() { @Override public void run() { try { GobblinServiceJobScheduler.this.orchestrator.getSpecCompiler().awaitHealthy(); } catch (InterruptedException e) { throw new RuntimeException(e); } scheduleSpecsFromCatalog(); } }); scheduleSpec.start(); } } else { for (Spec spec : this.scheduledFlowSpecs.values()) { onDeleteSpec(spec.getUri(), spec.getVersion()); } this.isActive = isActive; } } }
GobblinServiceJobScheduler extends JobScheduler implements SpecCatalogListener { public synchronized void setActive(boolean isActive) { if (this.isActive == isActive) { return; } if (isActive) { this.isActive = isActive; if (this.flowCatalog.isPresent()) { Thread scheduleSpec = new Thread(new Runnable() { @Override public void run() { try { GobblinServiceJobScheduler.this.orchestrator.getSpecCompiler().awaitHealthy(); } catch (InterruptedException e) { throw new RuntimeException(e); } scheduleSpecsFromCatalog(); } }); scheduleSpec.start(); } } else { for (Spec spec : this.scheduledFlowSpecs.values()) { onDeleteSpec(spec.getUri(), spec.getVersion()); } this.isActive = isActive; } } GobblinServiceJobScheduler(String serviceName, Config config, Optional<HelixManager> helixManager,
      Optional<FlowCatalog> flowCatalog, Optional<TopologyCatalog> topologyCatalog, Orchestrator orchestrator,
      SchedulerService schedulerService, Optional<Logger> log);  GobblinServiceJobScheduler(String serviceName, Config config, Optional<HelixManager> helixManager,
      Optional<FlowCatalog> flowCatalog, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager,
      SchedulerService schedulerService, Optional<Logger> log); }
GobblinServiceJobScheduler extends JobScheduler implements SpecCatalogListener { public synchronized void setActive(boolean isActive) { if (this.isActive == isActive) { return; } if (isActive) { this.isActive = isActive; if (this.flowCatalog.isPresent()) { Thread scheduleSpec = new Thread(new Runnable() { @Override public void run() { try { GobblinServiceJobScheduler.this.orchestrator.getSpecCompiler().awaitHealthy(); } catch (InterruptedException e) { throw new RuntimeException(e); } scheduleSpecsFromCatalog(); } }); scheduleSpec.start(); } } else { for (Spec spec : this.scheduledFlowSpecs.values()) { onDeleteSpec(spec.getUri(), spec.getVersion()); } this.isActive = isActive; } } GobblinServiceJobScheduler(String serviceName, Config config, Optional<HelixManager> helixManager,
      Optional<FlowCatalog> flowCatalog, Optional<TopologyCatalog> topologyCatalog, Orchestrator orchestrator,
      SchedulerService schedulerService, Optional<Logger> log);  GobblinServiceJobScheduler(String serviceName, Config config, Optional<HelixManager> helixManager,
      Optional<FlowCatalog> flowCatalog, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager,
      SchedulerService schedulerService, Optional<Logger> log); synchronized void setActive(boolean isActive); @Override synchronized void scheduleJob(Properties jobProps, JobListener jobListener); @Override void runJob(Properties jobProps, JobListener jobListener); @Override AddSpecResponse onAddSpec(Spec addedSpec); void onDeleteSpec(URI deletedSpecURI, String deletedSpecVersion); @Override void onDeleteSpec(URI deletedSpecURI, String deletedSpecVersion, Properties headers); @Override void onUpdateSpec(Spec updatedSpec); }
GobblinServiceJobScheduler extends JobScheduler implements SpecCatalogListener { public synchronized void setActive(boolean isActive) { if (this.isActive == isActive) { return; } if (isActive) { this.isActive = isActive; if (this.flowCatalog.isPresent()) { Thread scheduleSpec = new Thread(new Runnable() { @Override public void run() { try { GobblinServiceJobScheduler.this.orchestrator.getSpecCompiler().awaitHealthy(); } catch (InterruptedException e) { throw new RuntimeException(e); } scheduleSpecsFromCatalog(); } }); scheduleSpec.start(); } } else { for (Spec spec : this.scheduledFlowSpecs.values()) { onDeleteSpec(spec.getUri(), spec.getVersion()); } this.isActive = isActive; } } GobblinServiceJobScheduler(String serviceName, Config config, Optional<HelixManager> helixManager,
      Optional<FlowCatalog> flowCatalog, Optional<TopologyCatalog> topologyCatalog, Orchestrator orchestrator,
      SchedulerService schedulerService, Optional<Logger> log);  GobblinServiceJobScheduler(String serviceName, Config config, Optional<HelixManager> helixManager,
      Optional<FlowCatalog> flowCatalog, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager,
      SchedulerService schedulerService, Optional<Logger> log); synchronized void setActive(boolean isActive); @Override synchronized void scheduleJob(Properties jobProps, JobListener jobListener); @Override void runJob(Properties jobProps, JobListener jobListener); @Override AddSpecResponse onAddSpec(Spec addedSpec); void onDeleteSpec(URI deletedSpecURI, String deletedSpecVersion); @Override void onDeleteSpec(URI deletedSpecURI, String deletedSpecVersion, Properties headers); @Override void onUpdateSpec(Spec updatedSpec); static final String GOBBLIN_SERVICE_SCHEDULER_DR_NOMINATED; static final String DR_FILTER_TAG; }