Add Flink2 support for pipeline source connector.

Author: lvyanquan

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java

@@ -154,8 +154,7 @@ void testConsumingAllEvents(boolean parallelismSnapshot)
                         + " PRIMARY KEY (name) NOT ENFORCED"
                         + ") WITH ("
                         + " 'connector' = 'values',"
-                        + " 'sink-insert-only' = 'false',"
-                        + " 'sink-expected-messages-num' = '20'"
+                        + " 'sink-insert-only' = 'false'"
                         + ")";
         tEnv.executeSql(sourceDDL);
         tEnv.executeSql(sinkDDL);
@@ -187,7 +186,20 @@ void testConsumingAllEvents(boolean parallelismSnapshot)
             statement.execute("DELETE FROM inventory.products WHERE id=111;");
         }
 
-        waitForSinkSize("sink", 20);
+        // Wait for the final aggregated result
+        // Note: We use waitForSinkResult instead of waitForSinkSize because Flink 2.x
+        // optimizations may reduce the number of intermediate changelog messages
+        String[] expected =
+                new String[] {
+                    "scooter,3.140",
+                    "car battery,8.100",
+                    "12-pack drill bits,0.800",
+                    "hammer,2.625",
+                    "rocks,5.100",
+                    "jacket,0.600",
+                    "spare tire,22.200"
+                };
+        waitForSinkResult("sink", Arrays.asList(expected));
 
         /*
          * <pre>
@@ -215,17 +227,6 @@ void testConsumingAllEvents(boolean parallelismSnapshot)
          * </pre>
          */
 
-        String[] expected =
-                new String[] {
-                    "scooter,3.140",
-                    "car battery,8.100",
-                    "12-pack drill bits,0.800",
-                    "hammer,2.625",
-                    "rocks,5.100",
-                    "jacket,0.600",
-                    "spare tire,22.200"
-                };
-
         List<String> actual = TestValuesTableFactory.getResultsAsStrings("sink");
         Assertions.assertThat(actual).containsExactlyInAnyOrder(expected);
 
@@ -891,8 +892,7 @@ void testUpsertMode(boolean parallelismSnapshot) throws Exception {
                         + " PRIMARY KEY (name) NOT ENFORCED"
                         + ") WITH ("
                         + " 'connector' = 'values',"
-                        + " 'sink-insert-only' = 'false',"
-                        + " 'sink-expected-messages-num' = '20'"
+                        + " 'sink-insert-only' = 'false'"
                         + ")";
         tEnv.executeSql(sourceDDL);
         tEnv.executeSql(sinkDDL);
@@ -923,7 +923,20 @@ void testUpsertMode(boolean parallelismSnapshot) throws Exception {
             statement.execute("DELETE FROM inventory.products WHERE id=111;");
         }
 
-        waitForSinkSize("sink", 20);
+        // Wait for the final aggregated result
+        // Note: We use waitForSinkResult instead of waitForSinkSize because Flink 2.x
+        // optimizations may reduce the number of intermediate changelog messages
+        String[] expected =
+                new String[] {
+                    "scooter,3.140",
+                    "car battery,8.100",
+                    "12-pack drill bits,0.800",
+                    "hammer,2.625",
+                    "rocks,5.100",
+                    "jacket,0.600",
+                    "spare tire,22.200"
+                };
+        waitForSinkResult("sink", Arrays.asList(expected));
 
         /*
          * <pre>
@@ -951,17 +964,6 @@ void testUpsertMode(boolean parallelismSnapshot) throws Exception {
          * </pre>
          */
 
-        String[] expected =
-                new String[] {
-                    "scooter,3.140",
-                    "car battery,8.100",
-                    "12-pack drill bits,0.800",
-                    "hammer,2.625",
-                    "rocks,5.100",
-                    "jacket,0.600",
-                    "spare tire,22.200"
-                };
-
         List<String> actual = TestValuesTableFactory.getResultsAsStrings("sink");
         Assertions.assertThat(actual).containsExactlyInAnyOrder(expected);
 

flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/sink2/Sink.java

@@ -73,5 +73,13 @@ public interface InitContext extends org.apache.flink.api.connector.sink2.InitCo
         default <MetaT> Optional<Consumer<MetaT>> metadataConsumer() {
             return Optional.empty();
         }
+
+        default int getSubtaskId() {
+            return this.getTaskInfo().getIndexOfThisSubtask();
+        }
+
+        default int getNumberOfParallelSubtasks() {
+            return this.getTaskInfo().getNumberOfParallelSubtasks();
+        }
     }
 }