@Test (dependsOnMethods = "createTopologySpec") public void createFlowSpec() throws Exception { IdentityFlowToJobSpecCompiler specCompiler = (IdentityFlowToJobSpecCompiler) this.orchestrator.getSpecCompiler(); SpecExecutor sei = specCompiler.getTopologySpecMap().values().iterator().next().getSpecExecutor(); Collection<Spec> specs = flowCatalog.getSpecs(); logger.info("[Before Create] Number of specs: " + specs.size()); int i=0; for (Spec spec : specs) { FlowSpec flowSpec = (FlowSpec) spec; logger.info("[Before Create] Spec " + i++ + ": " + gson.toJson(flowSpec)); } Assert.assertTrue(specs.size() == 0, "Spec store should be empty before addition"); Assert.assertTrue(((List)(sei.getProducer().get().listSpecs().get())).size() == 0, "SpecProducer should not know about " + "any Flow before addition"); this.flowCatalog.put(flowSpec); specs = flowCatalog.getSpecs(); logger.info("[After Create] Number of specs: " + specs.size()); i = 0; for (Spec spec : specs) { flowSpec = (FlowSpec) spec; logger.info("[After Create] Spec " + i++ + ": " + gson.toJson(flowSpec)); } Assert.assertTrue(specs.size() == 1, "Spec store should contain 1 Spec after addition"); Assert.assertTrue(((List)(sei.getProducer().get().listSpecs().get())).size() == 0, "SpecProducer should contain 0 " + "Spec after addition"); }
@VisibleForTesting public SpecCompiler getSpecCompiler() { return this.specCompiler; }
Orchestrator implements SpecCatalogListener, Instrumentable { @VisibleForTesting public SpecCompiler getSpecCompiler() { return this.specCompiler; } }
Orchestrator implements SpecCatalogListener, Instrumentable { @VisibleForTesting public SpecCompiler getSpecCompiler() { return this.specCompiler; } Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Optional<Logger> log,
      boolean instrumentationEnabled);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Optional<Logger> log);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Logger log);  Orchestrator(Config config, Logger log);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog);  Orchestrator(Config config); }
Orchestrator implements SpecCatalogListener, Instrumentable { @VisibleForTesting public SpecCompiler getSpecCompiler() { return this.specCompiler; } Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Optional<Logger> log,
      boolean instrumentationEnabled);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Optional<Logger> log);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Logger log);  Orchestrator(Config config, Logger log);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog);  Orchestrator(Config config); @VisibleForTesting SpecCompiler getSpecCompiler(); @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); void orchestrate(Spec spec); void remove(Spec spec, Properties headers); @Nonnull @Override MetricContext getMetricContext(); @Override boolean isInstrumentationEnabled(); @Override List<Tag<?>> generateTags(State state); @Override void switchMetricContext(List<Tag<?>> tags); @Override void switchMetricContext(MetricContext context); }
Orchestrator implements SpecCatalogListener, Instrumentable { @VisibleForTesting public SpecCompiler getSpecCompiler() { return this.specCompiler; } Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Optional<Logger> log,
      boolean instrumentationEnabled);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Optional<Logger> log);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog, Optional<DagManager> dagManager, Logger log);  Orchestrator(Config config, Logger log);  Orchestrator(Config config, Optional<TopologyCatalog> topologyCatalog);  Orchestrator(Config config); @VisibleForTesting SpecCompiler getSpecCompiler(); @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); void orchestrate(Spec spec); void remove(Spec spec, Properties headers); @Nonnull @Override MetricContext getMetricContext(); @Override boolean isInstrumentationEnabled(); @Override List<Tag<?>> generateTags(State state); @Override void switchMetricContext(List<Tag<?>> tags); @Override void switchMetricContext(MetricContext context);  }