@Test (expectedExceptions = {JobSpecNotFoundException.class}) public void testFetchJobSpecs() throws ExecutionException, InterruptedException, URISyntaxException, JobSpecNotFoundException, IOException { String verb1 = SpecExecutor.Verb.ADD.name(); String version1 = "1"; addJobSpec(jobSpecUriString, version1, verb1); this.jobConfigurationManager.fetchJobSpecs(); JobSpec jobSpec = this._jobCatalog.getJobSpec(new URI(jobSpecUriString)); Assert.assertTrue(jobSpec != null); Assert.assertTrue(jobSpec.getVersion().equals(version1)); Assert.assertTrue(jobSpec.getUri().getPath().equals(jobSpecUriString)); Path fsSpecConsumerPath = new Path(fsSpecConsumerPathString); Assert.assertEquals(this.fs.listStatus(fsSpecConsumerPath, new HiddenFilter()).length, 0); Assert.assertEquals(newJobConfigArrivalEventCount, 1); Assert.assertEquals(updateJobConfigArrivalEventCount, 0); Assert.assertEquals(deleteJobConfigArrivalEventCount, 0); String verb2 = SpecExecutor.Verb.UPDATE.name(); String version2 = "2"; addJobSpec(jobSpecUriString, version2, verb2); this.jobConfigurationManager.fetchJobSpecs(); jobSpec = this._jobCatalog.getJobSpec(new URI(jobSpecUriString)); Assert.assertTrue(jobSpec != null); Assert.assertTrue(jobSpec.getVersion().equals(version2)); Assert.assertEquals(this.fs.listStatus(fsSpecConsumerPath, new HiddenFilter()).length, 0); Assert.assertEquals(newJobConfigArrivalEventCount, 1); Assert.assertEquals(updateJobConfigArrivalEventCount, 1); Assert.assertEquals(deleteJobConfigArrivalEventCount, 0); String verb3 = SpecExecutor.Verb.DELETE.name(); addJobSpec(jobSpecUriString, version2, verb3); this.jobConfigurationManager.fetchJobSpecs(); Assert.assertEquals(this.fs.listStatus(fsSpecConsumerPath, new HiddenFilter()).length, 0); this._jobCatalog.getJobSpec(new URI(jobSpecUriString)); Assert.assertEquals(newJobConfigArrivalEventCount, 1); Assert.assertEquals(updateJobConfigArrivalEventCount, 1); Assert.assertEquals(deleteJobConfigArrivalEventCount, 1); }
void fetchJobSpecs() throws ExecutionException, InterruptedException { List<Pair<SpecExecutor.Verb, JobSpec>> jobSpecs = (List<Pair<SpecExecutor.Verb, JobSpec>>) this.specConsumer.changedSpecs().get(); log.info("Fetched {} job specs", jobSpecs.size()); for (Pair<SpecExecutor.Verb, JobSpec> entry : jobSpecs) { JobSpec jobSpec = entry.getValue(); SpecExecutor.Verb verb = entry.getKey(); if (verb.equals(SpecExecutor.Verb.ADD)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postNewJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.UPDATE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postUpdateJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.DELETE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().remove(jobSpec.getUri()); } postDeleteJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } try { this.specConsumer.commit(jobSpec); } catch (IOException e) { log.error("Error when committing to FsSpecConsumer: ", e); } } }
FsJobConfigurationManager extends JobConfigurationManager { void fetchJobSpecs() throws ExecutionException, InterruptedException { List<Pair<SpecExecutor.Verb, JobSpec>> jobSpecs = (List<Pair<SpecExecutor.Verb, JobSpec>>) this.specConsumer.changedSpecs().get(); log.info("Fetched {} job specs", jobSpecs.size()); for (Pair<SpecExecutor.Verb, JobSpec> entry : jobSpecs) { JobSpec jobSpec = entry.getValue(); SpecExecutor.Verb verb = entry.getKey(); if (verb.equals(SpecExecutor.Verb.ADD)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postNewJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.UPDATE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postUpdateJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.DELETE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().remove(jobSpec.getUri()); } postDeleteJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } try { this.specConsumer.commit(jobSpec); } catch (IOException e) { log.error("Error when committing to FsSpecConsumer: ", e); } } } }
FsJobConfigurationManager extends JobConfigurationManager { void fetchJobSpecs() throws ExecutionException, InterruptedException { List<Pair<SpecExecutor.Verb, JobSpec>> jobSpecs = (List<Pair<SpecExecutor.Verb, JobSpec>>) this.specConsumer.changedSpecs().get(); log.info("Fetched {} job specs", jobSpecs.size()); for (Pair<SpecExecutor.Verb, JobSpec> entry : jobSpecs) { JobSpec jobSpec = entry.getValue(); SpecExecutor.Verb verb = entry.getKey(); if (verb.equals(SpecExecutor.Verb.ADD)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postNewJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.UPDATE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postUpdateJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.DELETE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().remove(jobSpec.getUri()); } postDeleteJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } try { this.specConsumer.commit(jobSpec); } catch (IOException e) { log.error("Error when committing to FsSpecConsumer: ", e); } } } FsJobConfigurationManager(EventBus eventBus, Config config, FileSystem fs);  FsJobConfigurationManager(EventBus eventBus, Config config, MutableJobCatalog jobCatalog, FileSystem fs); }
FsJobConfigurationManager extends JobConfigurationManager { void fetchJobSpecs() throws ExecutionException, InterruptedException { List<Pair<SpecExecutor.Verb, JobSpec>> jobSpecs = (List<Pair<SpecExecutor.Verb, JobSpec>>) this.specConsumer.changedSpecs().get(); log.info("Fetched {} job specs", jobSpecs.size()); for (Pair<SpecExecutor.Verb, JobSpec> entry : jobSpecs) { JobSpec jobSpec = entry.getValue(); SpecExecutor.Verb verb = entry.getKey(); if (verb.equals(SpecExecutor.Verb.ADD)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postNewJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.UPDATE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postUpdateJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.DELETE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().remove(jobSpec.getUri()); } postDeleteJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } try { this.specConsumer.commit(jobSpec); } catch (IOException e) { log.error("Error when committing to FsSpecConsumer: ", e); } } } FsJobConfigurationManager(EventBus eventBus, Config config, FileSystem fs);  FsJobConfigurationManager(EventBus eventBus, Config config, MutableJobCatalog jobCatalog, FileSystem fs);  }
FsJobConfigurationManager extends JobConfigurationManager { void fetchJobSpecs() throws ExecutionException, InterruptedException { List<Pair<SpecExecutor.Verb, JobSpec>> jobSpecs = (List<Pair<SpecExecutor.Verb, JobSpec>>) this.specConsumer.changedSpecs().get(); log.info("Fetched {} job specs", jobSpecs.size()); for (Pair<SpecExecutor.Verb, JobSpec> entry : jobSpecs) { JobSpec jobSpec = entry.getValue(); SpecExecutor.Verb verb = entry.getKey(); if (verb.equals(SpecExecutor.Verb.ADD)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postNewJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.UPDATE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().put(jobSpec); } postUpdateJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } else if (verb.equals(SpecExecutor.Verb.DELETE)) { if (this.jobCatalogOptional.isPresent()) { this.jobCatalogOptional.get().remove(jobSpec.getUri()); } postDeleteJobConfigArrival(jobSpec.getUri().toString(), jobSpec.getConfigAsProperties()); } try { this.specConsumer.commit(jobSpec); } catch (IOException e) { log.error("Error when committing to FsSpecConsumer: ", e); } } } FsJobConfigurationManager(EventBus eventBus, Config config, FileSystem fs);  FsJobConfigurationManager(EventBus eventBus, Config config, MutableJobCatalog jobCatalog, FileSystem fs);   }