update upsert kafka source as well

Author: klam-shop

docs/content.zh/docs/connectors/table/upsert-kafka.md

@@ -119,7 +119,7 @@ of all available metadata fields.
       <td>必选</td>
       <td style="word-wrap: break-word;">(none)</td>
       <td>String</td>
-      <td>用于读取和写入的 Kafka topic 名称。</td>
+      <td>当表用作 source 时读取数据的 topic 名，或当表用作 sink 时写入的 topic 名。它还支持通过分号分隔的 topic 列表，如 <code>'topic-1;topic-2'</code> 来作为 source 的 topic 列表。注意，“topic-pattern”和“topic”只能指定其中一个。对于 sink 来说，topic 名是写入数据的 topic。它还支持 sink 的 topic 列表。提供的 topic 列表被视为 `topic` 元数据列的有效值的允许列表。如果提供了列表，对于 sink 表，“topic”元数据列是可写的并且必须指定。</td>
     </tr>
     <tr>
       <td><h5>properties.bootstrap.servers</h5></td>

docs/content/docs/connectors/table/upsert-kafka.md

@@ -129,7 +129,7 @@ Connector Options
       <td>required</td>
       <td style="word-wrap: break-word;">(none)</td>
       <td>String</td>
-      <td>The Kafka topic name to read from and write to.</td>
+      <td>Topic name(s) to read data from when the table is used as source, or topics for writing when the table is used as sink. It also supports topic list for source by separating topic by semicolon like <code>'topic-1;topic-2'</code>. Note, only one of "topic-pattern" and "topic" can be specified. For sinks, the topic name is the topic to write data. It also supports topic list for sinks. The provided topic-list is treated as a allow list of valid values for the `topic` metadata column. If a list is provided, for sink table, 'topic' metadata column is writable and must be specified.</td>
     </tr>
     <tr>
       <td><h5>properties.bootstrap.servers</h5></td>

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactory.java

@@ -250,7 +250,6 @@ private static void validateSource(
             Format keyFormat,
             Format valueFormat,
             int[] primaryKeyIndexes) {
-        validateTopic(tableOptions);
         validateScanBoundedMode(tableOptions);
         validateFormat(keyFormat, valueFormat, tableOptions);
         validatePKConstraints(primaryKeyIndexes);
@@ -266,15 +265,6 @@ private static void validateSink(
         validateSinkBufferFlush(tableOptions);
     }
 
-    private static void validateTopic(ReadableConfig tableOptions) {
-        List<String> topic = tableOptions.get(TOPIC);
-        if (topic.size() > 1) {
-            throw new ValidationException(
-                    "The 'upsert-kafka' connector doesn't support topic list now. "
-                            + "Please use single topic as the value of the parameter 'topic'.");
-        }
-    }
-
     private static void validateFormat(
             Format keyFormat, Format valueFormat, ReadableConfig tableOptions) {
         if (!keyFormat.getChangelogMode().containsOnly(RowKind.INSERT)) {

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java

@@ -167,7 +167,39 @@ public void testTableSource() {
                         SOURCE_KEY_FIELDS,
                         SOURCE_VALUE_FIELDS,
                         null,
-                        SOURCE_TOPIC,
+                        Collections.singletonList(SOURCE_TOPIC),
+                        UPSERT_KAFKA_SOURCE_PROPERTIES);
+        assertThat(actualSource).isEqualTo(expectedSource);
+
+        final KafkaDynamicSource actualUpsertKafkaSource = (KafkaDynamicSource) actualSource;
+        ScanTableSource.ScanRuntimeProvider provider =
+                actualUpsertKafkaSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
+        assertKafkaSource(provider);
+    }
+
+    @Test
+    public void testTableSourceWithTopicList() {
+        final Map<String, String> modifiedOptions =
+                getModifiedOptions(
+                        getFullSourceOptions(),
+                        options -> {
+                            options.put(
+                                    "topic", String.format("%s;%s", SOURCE_TOPIC, SOURCE_TOPIC));
+                        });
+        final DataType producedDataType = SOURCE_SCHEMA.toPhysicalRowDataType();
+        // Construct table source using options and table source factory
+        final DynamicTableSource actualSource =
+                createTableSource(SOURCE_SCHEMA, modifiedOptions);
+
+        final KafkaDynamicSource expectedSource =
+                createExpectedScanSource(
+                        producedDataType,
+                        keyDecodingFormat,
+                        valueDecodingFormat,
+                        SOURCE_KEY_FIELDS,
+                        SOURCE_VALUE_FIELDS,
+                        null,
+                        Arrays.asList(SOURCE_TOPIC, SOURCE_TOPIC),
                         UPSERT_KAFKA_SOURCE_PROPERTIES);
         assertThat(actualSource).isEqualTo(expectedSource);
 
@@ -819,7 +851,7 @@ private KafkaDynamicSource createExpectedScanSource(
             int[] keyFields,
             int[] valueFields,
             String keyPrefix,
-            String topic,
+            List<String> topic,
             Properties properties) {
         return new KafkaDynamicSource(
                 producedDataType,
@@ -828,7 +860,7 @@ private KafkaDynamicSource createExpectedScanSource(
                 keyFields,
                 valueFields,
                 keyPrefix,
-                Collections.singletonList(topic),
+                topic,
                 null,
                 properties,
                 StartupMode.EARLIEST,