@Test public void testDisableFlowRunImmediatelyOnStart() throws Exception { Properties properties = new Properties(); properties.setProperty(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, "true"); properties.setProperty(ConfigurationKeys.JOB_SCHEDULE_KEY, TEST_SCHEDULE); properties.setProperty(ConfigurationKeys.JOB_GROUP_KEY, TEST_GROUP_NAME); properties.setProperty(ConfigurationKeys.JOB_NAME_KEY, TEST_FLOW_NAME); Config config = ConfigFactory.parseProperties(properties); FlowSpec spec = FlowSpec.builder().withTemplate(new URI(TEST_TEMPLATE_URI)).withVersion("version") .withConfigAsProperties(properties).withConfig(config).build(); FlowSpec modifiedSpec = (FlowSpec) GobblinServiceJobScheduler.disableFlowRunImmediatelyOnStart(spec); for (URI templateURI : modifiedSpec.getTemplateURIs().get()) { Assert.assertEquals(templateURI.toString(), TEST_TEMPLATE_URI); } Assert.assertEquals(modifiedSpec.getVersion(), "version"); Config modifiedConfig = modifiedSpec.getConfig(); Assert.assertFalse(modifiedConfig.getBoolean(ConfigurationKeys.FLOW_RUN_IMMEDIATELY)); Assert.assertEquals(modifiedConfig.getString(ConfigurationKeys.JOB_GROUP_KEY), TEST_GROUP_NAME); Assert.assertEquals(modifiedConfig.getString(ConfigurationKeys.JOB_NAME_KEY), TEST_FLOW_NAME); }
@VisibleForTesting protected static Spec disableFlowRunImmediatelyOnStart(FlowSpec spec) { Properties properties = spec.getConfigAsProperties(); properties.setProperty(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, "false"); Config config = ConfigFactory.parseProperties(properties); FlowSpec flowSpec = new FlowSpec(spec.getUri(), spec.getVersion(), spec.getDescription(), config, properties, spec.getTemplateURIs(), spec.getChildSpecs()); return flowSpec; }
GobblinServiceJobScheduler extends JobScheduler implements SpecCatalogListener { @VisibleForTesting protected static Spec disableFlowRunImmediatelyOnStart(FlowSpec spec) { Properties properties = spec.getConfigAsProperties(); properties.setProperty(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, "false"); Config config = ConfigFactory.parseProperties(properties); FlowSpec flowSpec = new FlowSpec(spec.getUri(), spec.getVersion(), spec.getDescription(), config, properties, spec.getTemplateURIs(), spec.getChildSpecs()); return flowSpec; } }
GobblinServiceJobScheduler extends JobScheduler implements SpecCatalogListener { @VisibleForTesting protected static Spec disableFlowRunImmediatelyOnStart(FlowSpec spec) { Properties properties = spec.getConfigAsProperties(); properties.setProperty(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, "false"); Config config = ConfigFactory.parseProperties(properties); FlowSpec flowSpec = new FlowSpec(spec.getUri(), spec.getVersion(), spec.getDescription(), config, properties, spec.getTemplateURIs(), spec.getChildSpecs()); return flowSpec; } 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 { @VisibleForTesting protected static Spec disableFlowRunImmediatelyOnStart(FlowSpec spec) { Properties properties = spec.getConfigAsProperties(); properties.setProperty(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, "false"); Config config = ConfigFactory.parseProperties(properties); FlowSpec flowSpec = new FlowSpec(spec.getUri(), spec.getVersion(), spec.getDescription(), config, properties, spec.getTemplateURIs(), spec.getChildSpecs()); return flowSpec; } 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 { @VisibleForTesting protected static Spec disableFlowRunImmediatelyOnStart(FlowSpec spec) { Properties properties = spec.getConfigAsProperties(); properties.setProperty(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, "false"); Config config = ConfigFactory.parseProperties(properties); FlowSpec flowSpec = new FlowSpec(spec.getUri(), spec.getVersion(), spec.getDescription(), config, properties, spec.getTemplateURIs(), spec.getChildSpecs()); return flowSpec; } 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; }