@Test public void testGetListOfTopicNamesByFilteringTag() throws Exception { Properties properties = new Properties(); properties.setProperty(GOBBLIN_CONFIG_TAGS_WHITELIST, "/tags/whitelist"); properties.setProperty(GOBBLIN_CONFIG_FILTER, "/data/tracking"); properties.setProperty(GOBBLIN_CONFIG_COMMONPATH, "/data/tracking"); List<String> result = ConfigStoreUtils .getListOfTopicNamesByFilteringTag(properties, configClient, Optional.absent(), configStoreUri, GOBBLIN_CONFIG_TAGS_WHITELIST); Assert.assertEquals(result.size(), 2); Assert.assertTrue(result.contains("Topic1")); Assert.assertTrue(result.contains("Topic2")); }
public static List<String> getListOfTopicNamesByFilteringTag(Properties properties, ConfigClient configClient, Optional<Config> runtimeConfig, String configStoreUri, String tagConfName) { Preconditions.checkArgument(properties.containsKey(GOBBLIN_CONFIG_FILTER), "Missing required property " + GOBBLIN_CONFIG_FILTER); String filterString = properties.getProperty(GOBBLIN_CONFIG_FILTER); Path tagUri = new Path("/"); try { tagUri = new Path(getUriFromPath(new Path(properties.getProperty(tagConfName)), configStoreUri)); } catch (URISyntaxException ue) { log.error("Cannot construct a Tag URI due to the exception:", ue); } List<String> taggedTopics = new ArrayList<>(); ConfigStoreUtils.getTopicsURIFromConfigStore(configClient, tagUri, filterString, runtimeConfig) .forEach(((URI u) -> taggedTopics.add(ConfigStoreUtils.getTopicNameFromURI(u)))); return taggedTopics; }
ConfigStoreUtils { public static List<String> getListOfTopicNamesByFilteringTag(Properties properties, ConfigClient configClient, Optional<Config> runtimeConfig, String configStoreUri, String tagConfName) { Preconditions.checkArgument(properties.containsKey(GOBBLIN_CONFIG_FILTER), "Missing required property " + GOBBLIN_CONFIG_FILTER); String filterString = properties.getProperty(GOBBLIN_CONFIG_FILTER); Path tagUri = new Path("/"); try { tagUri = new Path(getUriFromPath(new Path(properties.getProperty(tagConfName)), configStoreUri)); } catch (URISyntaxException ue) { log.error("Cannot construct a Tag URI due to the exception:", ue); } List<String> taggedTopics = new ArrayList<>(); ConfigStoreUtils.getTopicsURIFromConfigStore(configClient, tagUri, filterString, runtimeConfig) .forEach(((URI u) -> taggedTopics.add(ConfigStoreUtils.getTopicNameFromURI(u)))); return taggedTopics; } }
ConfigStoreUtils { public static List<String> getListOfTopicNamesByFilteringTag(Properties properties, ConfigClient configClient, Optional<Config> runtimeConfig, String configStoreUri, String tagConfName) { Preconditions.checkArgument(properties.containsKey(GOBBLIN_CONFIG_FILTER), "Missing required property " + GOBBLIN_CONFIG_FILTER); String filterString = properties.getProperty(GOBBLIN_CONFIG_FILTER); Path tagUri = new Path("/"); try { tagUri = new Path(getUriFromPath(new Path(properties.getProperty(tagConfName)), configStoreUri)); } catch (URISyntaxException ue) { log.error("Cannot construct a Tag URI due to the exception:", ue); } List<String> taggedTopics = new ArrayList<>(); ConfigStoreUtils.getTopicsURIFromConfigStore(configClient, tagUri, filterString, runtimeConfig) .forEach(((URI u) -> taggedTopics.add(ConfigStoreUtils.getTopicNameFromURI(u)))); return taggedTopics; }  }
ConfigStoreUtils { public static List<String> getListOfTopicNamesByFilteringTag(Properties properties, ConfigClient configClient, Optional<Config> runtimeConfig, String configStoreUri, String tagConfName) { Preconditions.checkArgument(properties.containsKey(GOBBLIN_CONFIG_FILTER), "Missing required property " + GOBBLIN_CONFIG_FILTER); String filterString = properties.getProperty(GOBBLIN_CONFIG_FILTER); Path tagUri = new Path("/"); try { tagUri = new Path(getUriFromPath(new Path(properties.getProperty(tagConfName)), configStoreUri)); } catch (URISyntaxException ue) { log.error("Cannot construct a Tag URI due to the exception:", ue); } List<String> taggedTopics = new ArrayList<>(); ConfigStoreUtils.getTopicsURIFromConfigStore(configClient, tagUri, filterString, runtimeConfig) .forEach(((URI u) -> taggedTopics.add(ConfigStoreUtils.getTopicNameFromURI(u)))); return taggedTopics; }  static Collection<URI> getTopicsURIFromConfigStore(ConfigClient configClient, Path tagUri, String filterString,
      Optional<Config> runtimeConfig); static Optional<String> getConfigStoreUri(Properties properties); static String getTopicNameFromURI(URI uri); static URI getUriStringForTopic(String topicName, String commonPath, String configStoreUri); static Optional<Config> getConfigForTopic(Properties properties, String topicKey, ConfigClient configClient); static Optional<Config> getConfigForTopicWithName(Properties properties, String topicName, ConfigClient configClient); static Config getConfig(ConfigClient client, URI u, Optional<Config> runtimeConfig); static List<KafkaTopic> getTopicsFromConfigStore(Properties properties, String configStoreUri,
      GobblinKafkaConsumerClient kafkaConsumerClient); static List<String> getListOfTopicNamesByFilteringTag(Properties properties, ConfigClient configClient,
      Optional<Config> runtimeConfig, String configStoreUri, String tagConfName); @Deprecated static void setTopicsFromConfigStore(Properties properties, Set<String> blacklist, Set<String> whitelist,
      final String _blacklistTopicKey, final String _whitelistTopicKey); }
ConfigStoreUtils { public static List<String> getListOfTopicNamesByFilteringTag(Properties properties, ConfigClient configClient, Optional<Config> runtimeConfig, String configStoreUri, String tagConfName) { Preconditions.checkArgument(properties.containsKey(GOBBLIN_CONFIG_FILTER), "Missing required property " + GOBBLIN_CONFIG_FILTER); String filterString = properties.getProperty(GOBBLIN_CONFIG_FILTER); Path tagUri = new Path("/"); try { tagUri = new Path(getUriFromPath(new Path(properties.getProperty(tagConfName)), configStoreUri)); } catch (URISyntaxException ue) { log.error("Cannot construct a Tag URI due to the exception:", ue); } List<String> taggedTopics = new ArrayList<>(); ConfigStoreUtils.getTopicsURIFromConfigStore(configClient, tagUri, filterString, runtimeConfig) .forEach(((URI u) -> taggedTopics.add(ConfigStoreUtils.getTopicNameFromURI(u)))); return taggedTopics; }  static Collection<URI> getTopicsURIFromConfigStore(ConfigClient configClient, Path tagUri, String filterString,
      Optional<Config> runtimeConfig); static Optional<String> getConfigStoreUri(Properties properties); static String getTopicNameFromURI(URI uri); static URI getUriStringForTopic(String topicName, String commonPath, String configStoreUri); static Optional<Config> getConfigForTopic(Properties properties, String topicKey, ConfigClient configClient); static Optional<Config> getConfigForTopicWithName(Properties properties, String topicName, ConfigClient configClient); static Config getConfig(ConfigClient client, URI u, Optional<Config> runtimeConfig); static List<KafkaTopic> getTopicsFromConfigStore(Properties properties, String configStoreUri,
      GobblinKafkaConsumerClient kafkaConsumerClient); static List<String> getListOfTopicNamesByFilteringTag(Properties properties, ConfigClient configClient,
      Optional<Config> runtimeConfig, String configStoreUri, String tagConfName); @Deprecated static void setTopicsFromConfigStore(Properties properties, Set<String> blacklist, Set<String> whitelist,
      final String _blacklistTopicKey, final String _whitelistTopicKey); static final String GOBBLIN_CONFIG_TAGS_WHITELIST; static final String GOBBLIN_CONFIG_TAGS_BLACKLIST; static final String GOBBLIN_CONFIG_FILTER; static final String GOBBLIN_CONFIG_COMMONPATH; }