@Test public void testGetTopologies() { Collection<TopologySpec> topologySpecs = _configBasedTopologySpecFactory.getTopologies(); Assert.assertTrue(topologySpecs.size() == 2, "Expected 2 topologies but received: " + topologySpecs.size()); Iterator<TopologySpec> topologySpecIterator = topologySpecs.iterator(); TopologySpec topologySpec1 = topologySpecIterator.next(); Assert.assertTrue(topologySpec1.getDescription().equals("Topology for cluster"), "Description did not match with construction"); Assert.assertTrue(topologySpec1.getVersion().equals("1"), "Version did not match with construction"); TopologySpec topologySpec2 = topologySpecIterator.next(); Assert.assertTrue(topologySpec2.getDescription().equals("Topology for Azkaban"), "Description did not match with construction"); Assert.assertTrue(topologySpec2.getVersion().equals("2"), "Version did not match with construction"); }
@Override public Collection<TopologySpec> getTopologies() { if (!_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)) { return Collections.EMPTY_LIST; } Collection<TopologySpec> topologySpecs = Lists.newArrayList(); Collection<String> topologyNames = SPLIT_BY_COMMA.splitToList( _config.getString(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)); for (String topologyName : topologyNames) { Preconditions.checkArgument(_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName), "Config does not contain Topology Factory descriptor for Topology " + topologyName); Config topologyConfig = _config.getConfig(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName); String description = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_DESCRIPTION_KEY, "NA"); String version = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_VERSION_KEY, "-1"); String specExecutorClass = ServiceConfigKeys.DEFAULT_SPEC_EXECUTOR; if (topologyConfig.hasPath(ServiceConfigKeys.SPEC_EXECUTOR_KEY)) { specExecutorClass = topologyConfig.getString(ServiceConfigKeys.SPEC_EXECUTOR_KEY); } SpecExecutor specExecutor; try { _log.info("Using SpecProducer class name/alias " + specExecutorClass); specExecutor = (SpecExecutor) ConstructorUtils .invokeConstructor(Class.forName(_aliasResolver .resolve(specExecutorClass)), topologyConfig); } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException | ClassNotFoundException e) { if (e.getCause() != null) { throw new RuntimeException(e.getCause()); } else { throw new RuntimeException(e); } } TopologySpec.Builder topologySpecBuilder = TopologySpec .builder(topologyConfig.getString(ServiceConfigKeys.TOPOLOGYSPEC_URI_KEY)) .withConfig(topologyConfig) .withDescription(description) .withVersion(version) .withSpecExecutor(specExecutor); topologySpecs.add(topologySpecBuilder.build()); } return topologySpecs; }
ConfigBasedTopologySpecFactory implements TopologySpecFactory { @Override public Collection<TopologySpec> getTopologies() { if (!_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)) { return Collections.EMPTY_LIST; } Collection<TopologySpec> topologySpecs = Lists.newArrayList(); Collection<String> topologyNames = SPLIT_BY_COMMA.splitToList( _config.getString(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)); for (String topologyName : topologyNames) { Preconditions.checkArgument(_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName), "Config does not contain Topology Factory descriptor for Topology " + topologyName); Config topologyConfig = _config.getConfig(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName); String description = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_DESCRIPTION_KEY, "NA"); String version = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_VERSION_KEY, "-1"); String specExecutorClass = ServiceConfigKeys.DEFAULT_SPEC_EXECUTOR; if (topologyConfig.hasPath(ServiceConfigKeys.SPEC_EXECUTOR_KEY)) { specExecutorClass = topologyConfig.getString(ServiceConfigKeys.SPEC_EXECUTOR_KEY); } SpecExecutor specExecutor; try { _log.info("Using SpecProducer class name/alias " + specExecutorClass); specExecutor = (SpecExecutor) ConstructorUtils .invokeConstructor(Class.forName(_aliasResolver .resolve(specExecutorClass)), topologyConfig); } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException | ClassNotFoundException e) { if (e.getCause() != null) { throw new RuntimeException(e.getCause()); } else { throw new RuntimeException(e); } } TopologySpec.Builder topologySpecBuilder = TopologySpec .builder(topologyConfig.getString(ServiceConfigKeys.TOPOLOGYSPEC_URI_KEY)) .withConfig(topologyConfig) .withDescription(description) .withVersion(version) .withSpecExecutor(specExecutor); topologySpecs.add(topologySpecBuilder.build()); } return topologySpecs; } }
ConfigBasedTopologySpecFactory implements TopologySpecFactory { @Override public Collection<TopologySpec> getTopologies() { if (!_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)) { return Collections.EMPTY_LIST; } Collection<TopologySpec> topologySpecs = Lists.newArrayList(); Collection<String> topologyNames = SPLIT_BY_COMMA.splitToList( _config.getString(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)); for (String topologyName : topologyNames) { Preconditions.checkArgument(_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName), "Config does not contain Topology Factory descriptor for Topology " + topologyName); Config topologyConfig = _config.getConfig(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName); String description = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_DESCRIPTION_KEY, "NA"); String version = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_VERSION_KEY, "-1"); String specExecutorClass = ServiceConfigKeys.DEFAULT_SPEC_EXECUTOR; if (topologyConfig.hasPath(ServiceConfigKeys.SPEC_EXECUTOR_KEY)) { specExecutorClass = topologyConfig.getString(ServiceConfigKeys.SPEC_EXECUTOR_KEY); } SpecExecutor specExecutor; try { _log.info("Using SpecProducer class name/alias " + specExecutorClass); specExecutor = (SpecExecutor) ConstructorUtils .invokeConstructor(Class.forName(_aliasResolver .resolve(specExecutorClass)), topologyConfig); } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException | ClassNotFoundException e) { if (e.getCause() != null) { throw new RuntimeException(e.getCause()); } else { throw new RuntimeException(e); } } TopologySpec.Builder topologySpecBuilder = TopologySpec .builder(topologyConfig.getString(ServiceConfigKeys.TOPOLOGYSPEC_URI_KEY)) .withConfig(topologyConfig) .withDescription(description) .withVersion(version) .withSpecExecutor(specExecutor); topologySpecs.add(topologySpecBuilder.build()); } return topologySpecs; } ConfigBasedTopologySpecFactory(Config config);  ConfigBasedTopologySpecFactory(Config config, Optional<Logger> log); }
ConfigBasedTopologySpecFactory implements TopologySpecFactory { @Override public Collection<TopologySpec> getTopologies() { if (!_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)) { return Collections.EMPTY_LIST; } Collection<TopologySpec> topologySpecs = Lists.newArrayList(); Collection<String> topologyNames = SPLIT_BY_COMMA.splitToList( _config.getString(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)); for (String topologyName : topologyNames) { Preconditions.checkArgument(_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName), "Config does not contain Topology Factory descriptor for Topology " + topologyName); Config topologyConfig = _config.getConfig(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName); String description = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_DESCRIPTION_KEY, "NA"); String version = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_VERSION_KEY, "-1"); String specExecutorClass = ServiceConfigKeys.DEFAULT_SPEC_EXECUTOR; if (topologyConfig.hasPath(ServiceConfigKeys.SPEC_EXECUTOR_KEY)) { specExecutorClass = topologyConfig.getString(ServiceConfigKeys.SPEC_EXECUTOR_KEY); } SpecExecutor specExecutor; try { _log.info("Using SpecProducer class name/alias " + specExecutorClass); specExecutor = (SpecExecutor) ConstructorUtils .invokeConstructor(Class.forName(_aliasResolver .resolve(specExecutorClass)), topologyConfig); } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException | ClassNotFoundException e) { if (e.getCause() != null) { throw new RuntimeException(e.getCause()); } else { throw new RuntimeException(e); } } TopologySpec.Builder topologySpecBuilder = TopologySpec .builder(topologyConfig.getString(ServiceConfigKeys.TOPOLOGYSPEC_URI_KEY)) .withConfig(topologyConfig) .withDescription(description) .withVersion(version) .withSpecExecutor(specExecutor); topologySpecs.add(topologySpecBuilder.build()); } return topologySpecs; } ConfigBasedTopologySpecFactory(Config config);  ConfigBasedTopologySpecFactory(Config config, Optional<Logger> log); @Override Collection<TopologySpec> getTopologies(); }
ConfigBasedTopologySpecFactory implements TopologySpecFactory { @Override public Collection<TopologySpec> getTopologies() { if (!_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)) { return Collections.EMPTY_LIST; } Collection<TopologySpec> topologySpecs = Lists.newArrayList(); Collection<String> topologyNames = SPLIT_BY_COMMA.splitToList( _config.getString(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY)); for (String topologyName : topologyNames) { Preconditions.checkArgument(_config.hasPath(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName), "Config does not contain Topology Factory descriptor for Topology " + topologyName); Config topologyConfig = _config.getConfig(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX + topologyName); String description = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_DESCRIPTION_KEY, "NA"); String version = ConfigUtils.getString(topologyConfig, ServiceConfigKeys.TOPOLOGYSPEC_VERSION_KEY, "-1"); String specExecutorClass = ServiceConfigKeys.DEFAULT_SPEC_EXECUTOR; if (topologyConfig.hasPath(ServiceConfigKeys.SPEC_EXECUTOR_KEY)) { specExecutorClass = topologyConfig.getString(ServiceConfigKeys.SPEC_EXECUTOR_KEY); } SpecExecutor specExecutor; try { _log.info("Using SpecProducer class name/alias " + specExecutorClass); specExecutor = (SpecExecutor) ConstructorUtils .invokeConstructor(Class.forName(_aliasResolver .resolve(specExecutorClass)), topologyConfig); } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException | ClassNotFoundException e) { if (e.getCause() != null) { throw new RuntimeException(e.getCause()); } else { throw new RuntimeException(e); } } TopologySpec.Builder topologySpecBuilder = TopologySpec .builder(topologyConfig.getString(ServiceConfigKeys.TOPOLOGYSPEC_URI_KEY)) .withConfig(topologyConfig) .withDescription(description) .withVersion(version) .withSpecExecutor(specExecutor); topologySpecs.add(topologySpecBuilder.build()); } return topologySpecs; } ConfigBasedTopologySpecFactory(Config config);  ConfigBasedTopologySpecFactory(Config config, Optional<Logger> log); @Override Collection<TopologySpec> getTopologies();  }