[GOBBLIN-1927] Add topic validation support in KafkaSource, and add TopicNameValidator

gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/KafkaSource.java

@@ -61,6 +61,7 @@
 import org.apache.gobblin.metrics.event.lineage.LineageInfo;
 import org.apache.gobblin.source.extractor.extract.EventBasedSource;
 import org.apache.gobblin.source.extractor.extract.kafka.workunit.packer.KafkaWorkUnitPacker;
+import org.apache.gobblin.source.extractor.extract.kafka.validator.TopicValidators;
 import org.apache.gobblin.source.extractor.limiter.LimiterConfigurationKeys;
 import org.apache.gobblin.source.workunit.Extract;
 import org.apache.gobblin.source.workunit.MultiWorkUnit;
@@ -218,7 +219,7 @@ public List<WorkUnit> getWorkunits(SourceState state) {
 
       this.kafkaConsumerClient.set(kafkaConsumerClientFactory.create(config));
 
-      List<KafkaTopic> topics = getFilteredTopics(state);
+      List<KafkaTopic> topics = getValidTopics(getFilteredTopics(state), state);
       this.topicsToProcess = topics.stream().map(KafkaTopic::getName).collect(toSet());
 
       for (String topic : this.topicsToProcess) {
@@ -802,6 +803,7 @@ private WorkUnit getWorkUnitForTopicPartition(KafkaPartition partition, Offsets
   protected List<KafkaTopic> getFilteredTopics(SourceState state) {
     List<Pattern> blacklist = DatasetFilterUtils.getPatternList(state, TOPIC_BLACKLIST);
     List<Pattern> whitelist = DatasetFilterUtils.getPatternList(state, TOPIC_WHITELIST);
+    // TODO: replace this with TopicNameValidator in the config once TopicValidators is rolled out.
     if (!state.getPropAsBoolean(KafkaSource.ALLOW_PERIOD_IN_TOPIC_NAME, true)) {
       blacklist.add(Pattern.compile(".*\\..*"));
     }
@@ -815,6 +817,13 @@ public void shutdown(SourceState state) {
     state.setProp(ConfigurationKeys.FAIL_TO_GET_OFFSET_COUNT, this.failToGetOffsetCount);
   }
 
+  /**
+   * Return topics that pass all the topic validators.
+   */
+  protected List<KafkaTopic> getValidTopics(List<KafkaTopic> topics, SourceState state) {
+    return new TopicValidators(state).validate(topics);
+  }
+
   /**
    * This class contains startOffset, earliestOffset and latestOffset for a Kafka partition.
    */

gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicNameValidator.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gobblin.source.extractor.extract.kafka.validator;
+
+import org.apache.gobblin.configuration.SourceState;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaTopic;
+
+/**
+ * A topic validator that validates the topic name
+ */
+public class TopicNameValidator extends TopicValidatorBase {
+  private static final String DOT = ".";
+
+  public TopicNameValidator(SourceState sourceState) {
+    super(sourceState);
+  }
+
+  /**
+   * Check if a topic name is valid, current rules are:
+   *  1. must not contain "."
+   * @param topic the topic to be validated
+   * @return true if the topic name is valid (aka. doesn't contain ".")
+   */
+  @Override
+  public boolean validate(KafkaTopic topic) {
+    return !topic.getName().contains(DOT);
+  }
+}

gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidatorBase.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gobblin.source.extractor.extract.kafka.validator;
+
+import org.apache.gobblin.configuration.SourceState;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaTopic;
+
+/**
+ * The base class of a topic validator
+ */
+public abstract class TopicValidatorBase {
+  protected SourceState sourceState;
+
+  public TopicValidatorBase(SourceState sourceState) {
+    this.sourceState = sourceState;
+  }
+
+  public abstract boolean validate(KafkaTopic topic);
+}

gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidators.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gobblin.source.extractor.extract.kafka.validator;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
+import org.apache.gobblin.configuration.SourceState;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaTopic;
+import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
+
+/**
+ * The TopicValidators contains a list of {@link TopicValidatorBase} that validate topics.
+ * To enable it, add below settings in the config:
+ *   gobblin.kafka.topicValidators=validator1_class_name,validator2_class_name...
+ */
+@Slf4j
+public class TopicValidators {
+  public static final String VALIDATOR_CLASSES_KEY = "gobblin.kafka.topicValidators";
+
+  public static final String VALIDATOR_CLASS_DELIMITER = ",";
+
+  private final List<TopicValidatorBase> validators = new ArrayList<>();
+
+  public TopicValidators(SourceState state) {
+    for (String validatorClassName : state.getPropAsList(VALIDATOR_CLASSES_KEY, StringUtils.EMPTY)) {
+      try {
+        this.validators.add(GobblinConstructorUtils.invokeConstructor(TopicValidatorBase.class, validatorClassName,
+            state));
+      } catch (Exception e) {
+        log.error("Failed to create topic validator: {}, due to {}", validatorClassName, e);
+      }
+    }
+  }
+
+  /**
+   * Validate topics with all the internal validators.
+   * Note: the validations for every topic run in parallel.
+   * @param topics the topics to be validated
+   * @return the topics that pass all the validators
+   */
+  public List<KafkaTopic> validate(List<KafkaTopic> topics) {
+    // Validate the topics in parallel
+    return topics.parallelStream()
+        .filter(this::validate)
+        .collect(Collectors.toList());
+  }
+
+  /**
+   * Validates a single topic with all the internal validators
+   */
+  private boolean validate(KafkaTopic topic) {
+    log.debug("Validating topic {} in thread: {}", topic, Thread.currentThread().getName());
+    for (TopicValidatorBase validator : this.validators) {
+      if (!validator.validate(topic)) {
+        log.info("Skip KafkaTopic: {}, by validator: {}", topic, validator.getClass().getName());
+        return false;
+      }
+    }
+    return true;
+  }
+}

gobblin-modules/gobblin-kafka-common/src/test/java/org/apache/gobblin/source/extractor/extract/kafka/KafkaSourceTest.java

@@ -25,6 +25,9 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.gobblin.source.extractor.extract.kafka.validator.TopicNameValidator;
+import org.apache.gobblin.source.extractor.extract.kafka.validator.TopicValidators;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -56,6 +59,36 @@ public void testGetFilteredTopics() {
     Assert.assertEquals(new TestKafkaSource(testKafkaClient).getFilteredTopics(state), toKafkaTopicList(allTopics.subList(0, 3)));
   }
 
+  @Test
+  public void testTopicValidators() {
+    TestKafkaClient testKafkaClient = new TestKafkaClient();
+    List<String> allTopics = Arrays.asList(
+        "Topic1", "topic-v2", "topic3", // allowed
+        "topic-with.period-in_middle", ".topic-with-period-at-start", "topicWithPeriodAtEnd.", //period topics
+        "not-allowed-topic");
+    testKafkaClient.testTopics = allTopics;
+    KafkaSource kafkaSource = new TestKafkaSource(testKafkaClient);
+
+    SourceState state = new SourceState();
+    state.setProp(KafkaSource.TOPIC_WHITELIST, ".*[Tt]opic.*");
+    state.setProp(KafkaSource.TOPIC_BLACKLIST, "not-allowed.*");
+    List<KafkaTopic> topicsToValidate = kafkaSource.getFilteredTopics(state);
+
+    // Test without TopicValidators in the state
+    Assert.assertTrue(CollectionUtils.isEqualCollection(kafkaSource.getValidTopics(topicsToValidate, state),
+        toKafkaTopicList(allTopics.subList(0, 6))));
+
+    // Test empty TopicValidators in the state
+    state.setProp(TopicValidators.VALIDATOR_CLASSES_KEY, "");
+    Assert.assertTrue(CollectionUtils.isEqualCollection(kafkaSource.getValidTopics(topicsToValidate, state),
+        toKafkaTopicList(allTopics.subList(0, 6))));
+
+    // Test TopicValidators with TopicNameValidator in the state
+    state.setProp(TopicValidators.VALIDATOR_CLASSES_KEY, TopicNameValidator.class.getName());
+    Assert.assertTrue(CollectionUtils.isEqualCollection(kafkaSource.getValidTopics(topicsToValidate, state),
+        toKafkaTopicList(allTopics.subList(0, 3))));
+  }
+
   public List<KafkaTopic> toKafkaTopicList(List<String> topicNames) {
     return topicNames.stream().map(topicName -> new KafkaTopic(topicName, Collections.emptyList())).collect(Collectors.toList());
   }

gobblin-modules/gobblin-kafka-common/src/test/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidatorsTest.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gobblin.source.extractor.extract.kafka.validator;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.gobblin.configuration.SourceState;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaTopic;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class TopicValidatorsTest {
+  @Test
+  public void testTopicValidators() {
+    List<String> allTopics = Arrays.asList(
+        "topic1", "topic2", // allowed
+        "topic-with.period-in_middle", ".topic-with-period-at-start", "topicWithPeriodAtEnd.", // bad topics
+        "topic3", "topic4"); // in deny list
+    List<KafkaTopic> topics = allTopics.stream()
+        .map(topicName -> new KafkaTopic(topicName, Collections.emptyList())).collect(Collectors.toList());
+
+    SourceState state = new SourceState();
+
+    // Without any topic validators
+    List<KafkaTopic> validTopics = new TopicValidators(state).validate(topics);
+    Assert.assertEquals(validTopics.size(), 7);
+
+    // Use 2 topic validators: TopicNameValidator and DenyListValidator
+    String validatorsToUse = String.join(TopicValidators.VALIDATOR_CLASS_DELIMITER,
+        ImmutableList.of(TopicNameValidator.class.getName(), DenyListValidator.class.getName()));
+    state.setProp(TopicValidators.VALIDATOR_CLASSES_KEY, validatorsToUse);
+    validTopics = new TopicValidators(state).validate(topics);
+
+    Assert.assertEquals(validTopics.size(), 2);
+    Assert.assertTrue(validTopics.stream().anyMatch(topic -> topic.getName().equals("topic1")));
+    Assert.assertTrue(validTopics.stream().anyMatch(topic -> topic.getName().equals("topic2")));
+  }
+
+  // A TopicValidator class to mimic a deny list
+  public static class DenyListValidator extends TopicValidatorBase {
+    Set<String> denyList = ImmutableSet.of("topic3", "topic4");
+
+    public DenyListValidator(SourceState sourceState) {
+      super(sourceState);
+    }
+
+    @Override
+    public boolean validate(KafkaTopic topic) {
+      return !this.denyList.contains(topic.getName());
+    }
+  }
+}