Check individual topics in ITCase

Author: klam-shop

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

@@ -200,29 +200,51 @@ public void testKafkaSourceSinkWithTopicList() throws Exception {
         // ---------- Produce an event time stream into Kafka -------------------
         String groupId = getStandardProps().getProperty("group.id");
         String bootstraps = getBootstrapServers();
-        final String createTable =
+        final String createTableTemplate =
+                "CREATE TABLE %s (\n"
+                        + "  `topic` STRING METADATA,\n"
+                        + "  `user_id` INT,\n"
+                        + "  `item_id` INT,\n"
+                        + "  `behavior` STRING\n"
+                        + ") WITH (\n"
+                        + "  'connector' = '%s',\n"
+                        + "  'topic' = '%s',\n"
+                        + "  'properties.bootstrap.servers' = '%s',\n"
+                        + "  'properties.group.id' = '%s',\n"
+                        + "  'scan.startup.mode' = 'earliest-offset',\n"
+                        + "  %s\n"
+                        + ")\n";
+        final String createTopicListTable =
                 String.format(
-                        "CREATE TABLE kafka (\n"
-                                + "  `topic` STRING METADATA,\n"
-                                + "  `user_id` INT,\n"
-                                + "  `item_id` INT,\n"
-                                + "  `behavior` STRING\n"
-                                + ") WITH (\n"
-                                + "  'connector' = '%s',\n"
-                                + "  'topic' = '%s;%s',\n"
-                                + "  'properties.bootstrap.servers' = '%s',\n"
-                                + "  'properties.group.id' = '%s',\n"
-                                + "  'scan.startup.mode' = 'earliest-offset',\n"
-                                + "  %s\n"
-                                + ")\n",
+                        "kafka",
+                        createTableTemplate,
+                        KafkaDynamicTableFactory.IDENTIFIER,
+                        String.join(";", Arrays.asList(topic1, topic2)),
+                        bootstraps,
+                        groupId,
+                        formatOptions());
+        final String createTopic1Table =
+                String.format(
+                        "topic1",
+                        createTableTemplate,
                         KafkaDynamicTableFactory.IDENTIFIER,
                         topic1,
+                        bootstraps,
+                        groupId,
+                        formatOptions());
+        final String createTopic2Table =
+                String.format(
+                        "topic2",
+                        createTableTemplate,
+                        KafkaDynamicTableFactory.IDENTIFIER,
                         topic2,
                         bootstraps,
                         groupId,
                         formatOptions());
 
-        tEnv.executeSql(createTable);
+        tEnv.executeSql(createTopicListTable);
+        tEnv.executeSql(createTopic1Table);
+        tEnv.executeSql(createTopic2Table);
 
         List<Row> values =
                 Arrays.asList(
@@ -233,14 +255,20 @@ public void testKafkaSourceSinkWithTopicList() throws Exception {
         // ---------- Consume stream from Kafka -------------------
 
         List<Row> results = collectAllRows(tEnv.sqlQuery("SELECT * from kafka"));
-
+        List<Row> topic1Results = collectAllRows(tEnv.sqlQuery("SELECT * from topic1"));
+        List<Row> topic2Results = collectAllRows(tEnv.sqlQuery("SELECT * from topic2"));
         assertThat(results)
-                .containsExactly(
+                .containsExactlyInAnyOrder(
                         Row.of(topic1, 1, 1102, "behavior 1"),
                         Row.of(topic2, 2, 1103, "behavior 2"));
+        assertThat(topic1Results)
+                .containsExactly(
+                        Row.of(topic1, 1, 1102, "behavior 1"));
+        assertThat(topic2Results)
+                .containsExactly(
+                        Row.of(topic2, 2, 1103, "behavior 2"));
 
         // ------------- cleanup -------------------
-
         deleteTestTopic(topic1);
         deleteTestTopic(topic2);
     }

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

@@ -188,8 +188,7 @@ public void testTableSourceWithTopicList() {
                         });
         final DataType producedDataType = SOURCE_SCHEMA.toPhysicalRowDataType();
         // Construct table source using options and table source factory
-        final DynamicTableSource actualSource =
-                createTableSource(SOURCE_SCHEMA, modifiedOptions);
+        final DynamicTableSource actualSource = createTableSource(SOURCE_SCHEMA, modifiedOptions);
 
         final KafkaDynamicSource expectedSource =
                 createExpectedScanSource(