@Test public void testGetTopicsFromConfigStore() throws Exception { KafkaTopic topic1 = new KafkaTopic("Topic1", Lists.newArrayList()); KafkaTopic topic2 = new KafkaTopic("Topic2", Lists.newArrayList()); KafkaTopic topic3 = new KafkaTopic("Topic3", Lists.newArrayList()); Mockito.when(mockClient.getFilteredTopics(anyList(), anyList())) .thenReturn(ImmutableList.of(topic1, topic2, topic3)); Properties properties = new Properties(); List<KafkaTopic> result = ConfigStoreUtils.getTopicsFromConfigStore(properties, configStoreUri, mockClient); Assert.assertEquals(result.size(), 3); properties.setProperty(GOBBLIN_CONFIG_TAGS_WHITELIST, "/tags/whitelist"); properties.setProperty(GOBBLIN_CONFIG_FILTER, "/data/tracking"); properties.setProperty(GOBBLIN_CONFIG_COMMONPATH, "/data/tracking"); result = ConfigStoreUtils.getTopicsFromConfigStore(properties, configStoreUri, mockClient); Assert.assertEquals(result.size(), 2); List<String> resultInString = result.stream().map(KafkaTopic::getName).collect(Collectors.toList()); Assert.assertTrue(resultInString.contains("Topic1")); Assert.assertTrue(resultInString.contains("Topic2")); properties.remove(GOBBLIN_CONFIG_TAGS_WHITELIST); properties.setProperty(GOBBLIN_CONFIG_TAGS_BLACKLIST, "/tags/blacklist"); result = ConfigStoreUtils.getTopicsFromConfigStore(properties, configStoreUri, mockClient); Assert.assertEquals(result.size(), 1); Assert.assertEquals(result.get(0).getName(), "Topic3"); }
public static List<KafkaTopic> getTopicsFromConfigStore(Properties properties, String configStoreUri, GobblinKafkaConsumerClient kafkaConsumerClient) { ConfigClient configClient = ConfigClient.createConfigClient(VersionStabilityPolicy.WEAK_LOCAL_STABILITY); State state = new State(); state.setProp(KafkaSource.TOPIC_WHITELIST, ".*"); state.setProp(KafkaSource.TOPIC_BLACKLIST, StringUtils.EMPTY); List<KafkaTopic> allTopics = kafkaConsumerClient.getFilteredTopics(DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_BLACKLIST), DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_WHITELIST)); Optional<Config> runtimeConfig = ConfigClientUtils.getOptionalRuntimeConfig(properties); if (properties.containsKey(GOBBLIN_CONFIG_TAGS_WHITELIST)) { List<String> whitelistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_WHITELIST); return allTopics.stream() .filter((KafkaTopic p) -> whitelistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else if (properties.containsKey(GOBBLIN_CONFIG_TAGS_BLACKLIST)) { List<String> blacklistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_BLACKLIST); return allTopics.stream() .filter((KafkaTopic p) -> !blacklistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else { log.warn("None of the blacklist or whitelist tags are provided"); return allTopics; } }
ConfigStoreUtils { public static List<KafkaTopic> getTopicsFromConfigStore(Properties properties, String configStoreUri, GobblinKafkaConsumerClient kafkaConsumerClient) { ConfigClient configClient = ConfigClient.createConfigClient(VersionStabilityPolicy.WEAK_LOCAL_STABILITY); State state = new State(); state.setProp(KafkaSource.TOPIC_WHITELIST, ".*"); state.setProp(KafkaSource.TOPIC_BLACKLIST, StringUtils.EMPTY); List<KafkaTopic> allTopics = kafkaConsumerClient.getFilteredTopics(DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_BLACKLIST), DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_WHITELIST)); Optional<Config> runtimeConfig = ConfigClientUtils.getOptionalRuntimeConfig(properties); if (properties.containsKey(GOBBLIN_CONFIG_TAGS_WHITELIST)) { List<String> whitelistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_WHITELIST); return allTopics.stream() .filter((KafkaTopic p) -> whitelistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else if (properties.containsKey(GOBBLIN_CONFIG_TAGS_BLACKLIST)) { List<String> blacklistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_BLACKLIST); return allTopics.stream() .filter((KafkaTopic p) -> !blacklistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else { log.warn("None of the blacklist or whitelist tags are provided"); return allTopics; } } }
ConfigStoreUtils { public static List<KafkaTopic> getTopicsFromConfigStore(Properties properties, String configStoreUri, GobblinKafkaConsumerClient kafkaConsumerClient) { ConfigClient configClient = ConfigClient.createConfigClient(VersionStabilityPolicy.WEAK_LOCAL_STABILITY); State state = new State(); state.setProp(KafkaSource.TOPIC_WHITELIST, ".*"); state.setProp(KafkaSource.TOPIC_BLACKLIST, StringUtils.EMPTY); List<KafkaTopic> allTopics = kafkaConsumerClient.getFilteredTopics(DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_BLACKLIST), DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_WHITELIST)); Optional<Config> runtimeConfig = ConfigClientUtils.getOptionalRuntimeConfig(properties); if (properties.containsKey(GOBBLIN_CONFIG_TAGS_WHITELIST)) { List<String> whitelistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_WHITELIST); return allTopics.stream() .filter((KafkaTopic p) -> whitelistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else if (properties.containsKey(GOBBLIN_CONFIG_TAGS_BLACKLIST)) { List<String> blacklistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_BLACKLIST); return allTopics.stream() .filter((KafkaTopic p) -> !blacklistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else { log.warn("None of the blacklist or whitelist tags are provided"); return allTopics; } }  }
ConfigStoreUtils { public static List<KafkaTopic> getTopicsFromConfigStore(Properties properties, String configStoreUri, GobblinKafkaConsumerClient kafkaConsumerClient) { ConfigClient configClient = ConfigClient.createConfigClient(VersionStabilityPolicy.WEAK_LOCAL_STABILITY); State state = new State(); state.setProp(KafkaSource.TOPIC_WHITELIST, ".*"); state.setProp(KafkaSource.TOPIC_BLACKLIST, StringUtils.EMPTY); List<KafkaTopic> allTopics = kafkaConsumerClient.getFilteredTopics(DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_BLACKLIST), DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_WHITELIST)); Optional<Config> runtimeConfig = ConfigClientUtils.getOptionalRuntimeConfig(properties); if (properties.containsKey(GOBBLIN_CONFIG_TAGS_WHITELIST)) { List<String> whitelistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_WHITELIST); return allTopics.stream() .filter((KafkaTopic p) -> whitelistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else if (properties.containsKey(GOBBLIN_CONFIG_TAGS_BLACKLIST)) { List<String> blacklistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_BLACKLIST); return allTopics.stream() .filter((KafkaTopic p) -> !blacklistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else { log.warn("None of the blacklist or whitelist tags are provided"); return allTopics; } }  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<KafkaTopic> getTopicsFromConfigStore(Properties properties, String configStoreUri, GobblinKafkaConsumerClient kafkaConsumerClient) { ConfigClient configClient = ConfigClient.createConfigClient(VersionStabilityPolicy.WEAK_LOCAL_STABILITY); State state = new State(); state.setProp(KafkaSource.TOPIC_WHITELIST, ".*"); state.setProp(KafkaSource.TOPIC_BLACKLIST, StringUtils.EMPTY); List<KafkaTopic> allTopics = kafkaConsumerClient.getFilteredTopics(DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_BLACKLIST), DatasetFilterUtils.getPatternList(state, KafkaSource.TOPIC_WHITELIST)); Optional<Config> runtimeConfig = ConfigClientUtils.getOptionalRuntimeConfig(properties); if (properties.containsKey(GOBBLIN_CONFIG_TAGS_WHITELIST)) { List<String> whitelistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_WHITELIST); return allTopics.stream() .filter((KafkaTopic p) -> whitelistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else if (properties.containsKey(GOBBLIN_CONFIG_TAGS_BLACKLIST)) { List<String> blacklistedTopics = getListOfTopicNamesByFilteringTag(properties, configClient, runtimeConfig, configStoreUri, GOBBLIN_CONFIG_TAGS_BLACKLIST); return allTopics.stream() .filter((KafkaTopic p) -> !blacklistedTopics.contains(p.getName())) .collect(Collectors.toList()); } else { log.warn("None of the blacklist or whitelist tags are provided"); return allTopics; } }  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; }