From adbc3089c39d9c1730473718c15f9fd7acca6348 Mon Sep 17 00:00:00 2001
From: Tao Qin <tqin@linkedin.com>
Date: Fri, 13 Oct 2023 00:00:25 -0700
Subject: [PATCH] Refine to address comments

---
 .../kafka/validator/TopicValidators.java       | 18 ++++++------------
 .../kafka/validator/TopicValidatorsTest.java   | 13 +++++++++----
 2 files changed, 15 insertions(+), 16 deletions(-)

diff --git a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidators.java b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidators.java
index 022493bb63d..19d1dc45953 100644
--- a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidators.java
+++ b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidators.java
@@ -16,12 +16,11 @@
  */
 package org.apache.gobblin.source.extractor.extract.kafka.validator;
 
-import com.google.common.base.Strings;
 import java.util.ArrayList;
-import java.util.Arrays;
 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;
@@ -39,19 +38,14 @@ public class TopicValidators {
   private final List<TopicValidatorBase> validators = new ArrayList<>();
 
   public TopicValidators(SourceState state) {
-    String validatorClasses = state.getProp(VALIDATOR_CLASSES_KEY);
-    if (Strings.isNullOrEmpty(validatorClasses)) {
-      return;
-    }
-
-    String[] validatorClassNames = validatorClasses.split(VALIDATOR_CLASS_DELIMITER);
-    Arrays.stream(validatorClassNames).forEach(validator -> {
+    for (String validatorClassName : state.getPropAsList(VALIDATOR_CLASSES_KEY, StringUtils.EMPTY)) {
       try {
-        this.validators.add(GobblinConstructorUtils.invokeConstructor(TopicValidatorBase.class, validator, state));
+        this.validators.add(GobblinConstructorUtils.invokeConstructor(TopicValidatorBase.class, validatorClassName,
+            state));
       } catch (Exception e) {
-        log.error("Failed to create topic validator: {}, due to {}", validator, e);
+        log.error("Failed to create topic validator: {}, due to {}", validatorClassName, e);
       }
-    });
+    }
   }
 
   /**
diff --git a/gobblin-modules/gobblin-kafka-common/src/test/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidatorsTest.java b/gobblin-modules/gobblin-kafka-common/src/test/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidatorsTest.java
index d2250f832ec..d75fdc6662e 100644
--- a/gobblin-modules/gobblin-kafka-common/src/test/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidatorsTest.java
+++ b/gobblin-modules/gobblin-kafka-common/src/test/java/org/apache/gobblin/source/extractor/extract/kafka/validator/TopicValidatorsTest.java
@@ -31,7 +31,7 @@
 
 public class TopicValidatorsTest {
   @Test
-  public void testTopicNameValidator() {
+  public void testTopicValidators() {
     List<String> allTopics = Arrays.asList(
         "topic1", "topic2", // allowed
         "topic-with.period-in_middle", ".topic-with-period-at-start", "topicWithPeriodAtEnd.", // bad topics
@@ -39,12 +39,17 @@ public void testTopicNameValidator() {
     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()));
-
-    SourceState state = new SourceState();
     state.setProp(TopicValidators.VALIDATOR_CLASSES_KEY, validatorsToUse);
-    List<KafkaTopic> validTopics = new TopicValidators(state).validate(topics);
+    validTopics = new TopicValidators(state).validate(topics);
 
     Assert.assertEquals(validTopics.size(), 2);
     Assert.assertTrue(validTopics.stream().anyMatch(topic -> topic.getName().equals("topic1")));