[INLONG-8695][Sort] rename topics to topic

apache · Sep 27, 2023 · 01f8878 · 01f8878
1 parent c79a7c0
commit 01f8878
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 10 deletions.
diff --git a/...ort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableFactory.java b/...ort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableFactory.java
@@ -71,7 +71,7 @@
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_SUBSCRIPTION_NAME;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
-import static org.apache.inlong.sort.pulsar.PulsarTableOptions.TOPICS;
+import static org.apache.inlong.sort.pulsar.PulsarTableOptions.TOPIC;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.VALUE_FORMAT;
 import static org.apache.inlong.sort.pulsar.PulsarTableValidationUtils.validatePrimaryKeyConstraints;
 import static org.apache.inlong.sort.pulsar.PulsarTableValidationUtils.validateTableSourceOptions;
@@ -169,7 +169,7 @@ public String factoryIdentifier() {
 
     @Override
     public Set<ConfigOption<?>> requiredOptions() {
-        return Stream.of(TOPICS, ADMIN_URL, SERVICE_URL).collect(Collectors.toSet());
+        return Stream.of(TOPIC, ADMIN_URL, SERVICE_URL).collect(Collectors.toSet());
     }
 
     @Override
@@ -198,7 +198,7 @@ public Set<ConfigOption<?>> optionalOptions() {
     @Override
     public Set<ConfigOption<?>> forwardOptions() {
         return Stream.of(
-                TOPICS,
+                TOPIC,
                 ADMIN_URL,
                 SERVICE_URL,
                 SOURCE_SUBSCRIPTION_TYPE,

diff --git a/...connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableOptionUtils.java b/...connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableOptionUtils.java
@@ -52,7 +52,7 @@
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
-import static org.apache.inlong.sort.pulsar.PulsarTableOptions.TOPICS;
+import static org.apache.inlong.sort.pulsar.PulsarTableOptions.TOPIC;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.VALUE_FORMAT;
 
 /**
@@ -151,7 +151,7 @@ public static int[] createValueFormatProjection(
     // --------------------------------------------------------------------------------------------
 
     public static List<String> getTopicListFromOptions(ReadableConfig tableOptions) {
-        return tableOptions.get(TOPICS);
+        return tableOptions.get(TOPIC);
     }
 
     public static Properties getPulsarProperties(ReadableConfig tableOptions) {

diff --git a/...ort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableOptions.java b/...ort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableOptions.java
@@ -45,8 +45,8 @@ public final class PulsarTableOptions {
     private PulsarTableOptions() {
     }
 
-    public static final ConfigOption<List<String>> TOPICS =
-            ConfigOptions.key("topics")
+    public static final ConfigOption<List<String>> TOPIC =
+            ConfigOptions.key("topic")
                     .stringType()
                     .asList()
                     .noDefaultValue()

diff --git a/...ectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableValidationUtils.java b/...ectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableValidationUtils.java
@@ -42,7 +42,7 @@
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
 import static org.apache.inlong.sort.pulsar.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
-import static org.apache.inlong.sort.pulsar.PulsarTableOptions.TOPICS;
+import static org.apache.inlong.sort.pulsar.PulsarTableOptions.TOPIC;
 import static org.apache.pulsar.common.naming.TopicName.isValid;
 
 /** Util class for source and sink validation rules.
@@ -78,11 +78,11 @@ public static void validateTableSourceOptions(ReadableConfig tableOptions) {
     }
 
     protected static void validateTopicsConfigs(ReadableConfig tableOptions) {
-        if (tableOptions.get(TOPICS).isEmpty()) {
+        if (tableOptions.get(TOPIC).isEmpty()) {
             throw new ValidationException("The topics list should not be empty.");
         }
 
-        for (String topic : tableOptions.get(TOPICS)) {
+        for (String topic : tableOptions.get(TOPIC)) {
             if (!isValid(topic)) {
                 throw new ValidationException(
                         String.format("The topics name %s is not a valid topic name.", topic));