@Test public void testModifyFlowTemplate() throws Exception { ObservingFSFlowEdgeTemplateCatalog catalog = new ObservingFSFlowEdgeTemplateCatalog(this.templateCatalogCfg, new ReentrantReadWriteLock()); ServiceManager serviceManager = new ServiceManager(Lists.newArrayList(catalog)); serviceManager.startAsync().awaitHealthy(5, TimeUnit.SECONDS); FlowTemplate flowTemplate1 = catalog.getFlowTemplate(new URI(FSFlowTemplateCatalogTest.TEST_TEMPLATE_DIR_URI)); FlowTemplate flowTemplate2 = catalog.getFlowTemplate(new URI(FSFlowTemplateCatalogTest.TEST_TEMPLATE_DIR_URI)); Assert.assertSame(flowTemplate1, flowTemplate2); Path flowConfPath = new File(new File(this.templateDir, FSFlowTemplateCatalogTest.TEST_TEMPLATE_NAME), "flow.conf").toPath(); List<String> lines = java.nio.file.Files.readAllLines(flowConfPath); for (int i = 0; i < lines.size(); i++) { if (lines.get(i).equals("gobblin.flow.edge.input.dataset.descriptor.0.format=avro")) { lines.set(i, "gobblin.flow.edge.input.dataset.descriptor.0.format=any"); break; } } java.nio.file.Files.write(flowConfPath, lines); Function testFunction = new GetFlowTemplateConfigFunction(new URI(FSFlowTemplateCatalogTest.TEST_TEMPLATE_DIR_URI), catalog, "gobblin.flow.edge.input.dataset.descriptor.0.format"); AssertWithBackoff.create().timeoutMs(10000).assertEquals(testFunction, "any", "flow template updated"); }
@Override public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { FlowTemplate flowTemplate = flowTemplateMap.getOrDefault(flowTemplateDirURI, null); if (flowTemplate == null) { flowTemplate = super.getFlowTemplate(flowTemplateDirURI); flowTemplateMap.put(flowTemplateDirURI, flowTemplate); } return flowTemplate; }
ObservingFSFlowEdgeTemplateCatalog extends FSFlowTemplateCatalog { @Override public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { FlowTemplate flowTemplate = flowTemplateMap.getOrDefault(flowTemplateDirURI, null); if (flowTemplate == null) { flowTemplate = super.getFlowTemplate(flowTemplateDirURI); flowTemplateMap.put(flowTemplateDirURI, flowTemplate); } return flowTemplate; } }
ObservingFSFlowEdgeTemplateCatalog extends FSFlowTemplateCatalog { @Override public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { FlowTemplate flowTemplate = flowTemplateMap.getOrDefault(flowTemplateDirURI, null); if (flowTemplate == null) { flowTemplate = super.getFlowTemplate(flowTemplateDirURI); flowTemplateMap.put(flowTemplateDirURI, flowTemplate); } return flowTemplate; } ObservingFSFlowEdgeTemplateCatalog(Config sysConfig, ReadWriteLock rwLock); }
ObservingFSFlowEdgeTemplateCatalog extends FSFlowTemplateCatalog { @Override public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { FlowTemplate flowTemplate = flowTemplateMap.getOrDefault(flowTemplateDirURI, null); if (flowTemplate == null) { flowTemplate = super.getFlowTemplate(flowTemplateDirURI); flowTemplateMap.put(flowTemplateDirURI, flowTemplate); } return flowTemplate; } ObservingFSFlowEdgeTemplateCatalog(Config sysConfig, ReadWriteLock rwLock); @Override FlowTemplate getFlowTemplate(URI flowTemplateDirURI); @Override List<JobTemplate> getJobTemplatesForFlow(URI flowTemplateDirURI); @Override boolean getAndSetShouldRefreshFlowGraph(boolean value); }
ObservingFSFlowEdgeTemplateCatalog extends FSFlowTemplateCatalog { @Override public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { FlowTemplate flowTemplate = flowTemplateMap.getOrDefault(flowTemplateDirURI, null); if (flowTemplate == null) { flowTemplate = super.getFlowTemplate(flowTemplateDirURI); flowTemplateMap.put(flowTemplateDirURI, flowTemplate); } return flowTemplate; } ObservingFSFlowEdgeTemplateCatalog(Config sysConfig, ReadWriteLock rwLock); @Override FlowTemplate getFlowTemplate(URI flowTemplateDirURI); @Override List<JobTemplate> getJobTemplatesForFlow(URI flowTemplateDirURI); @Override boolean getAndSetShouldRefreshFlowGraph(boolean value);  }