[FLINK-34941][checkpoint] Add CollectIteratorAssert#matchesRecordsFromSource and SinkTestSuiteBase#checkResultWithSemantic with former signature

Author: snuyanzin

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/CheckpointingMode.java

@@ -77,5 +77,17 @@ public enum CheckpointingMode {
      * scenarios, where a sustained very-low latency (such as few milliseconds) is needed, and where
      * occasional duplicate messages (on recovery) do not matter.
      */
-    AT_LEAST_ONCE
+    AT_LEAST_ONCE;
+
+    public static org.apache.flink.core.execution.CheckpointingMode convertToCheckpointingMode(
+            org.apache.flink.streaming.api.CheckpointingMode semantic) {
+        switch (semantic) {
+            case EXACTLY_ONCE:
+                return org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE;
+            case AT_LEAST_ONCE:
+                return org.apache.flink.core.execution.CheckpointingMode.AT_LEAST_ONCE;
+            default:
+                throw new IllegalArgumentException("Unsupported semantic: " + semantic);
+        }
+    }
 }

flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java

@@ -87,6 +87,7 @@
 import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
 import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForJobStatus;
 import static org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
+import static org.apache.flink.streaming.api.CheckpointingMode.convertToCheckpointingMode;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
@@ -513,6 +514,20 @@ protected void checkResultWithSemantic(
                 });
     }
 
+    /**
+     * This method is required for downstream projects e.g. Flink connectors extending this test for
+     * the case when there should be supported Flink versions below 1.20. Could be removed together
+     * with dropping support for Flink 1.19.
+     */
+    @Deprecated
+    protected void checkResultWithSemantic(
+            ExternalSystemDataReader<T> reader,
+            List<T> testData,
+            org.apache.flink.streaming.api.CheckpointingMode semantic)
+            throws Exception {
+        checkResultWithSemantic(reader, testData, convertToCheckpointingMode(semantic));
+    }
+
     /** Compare the metrics. */
     private boolean compareSinkMetrics(
             MetricQuerier metricQuerier,

flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssert.java

@@ -27,6 +27,8 @@
 import java.util.LinkedList;
 import java.util.List;
 
+import static org.apache.flink.streaming.api.CheckpointingMode.convertToCheckpointingMode;
+
 /**
  * This assertion used to compare records in the collect iterator to the target test data with
  * different semantic(AT_LEAST_ONCE, EXACTLY_ONCE).
@@ -51,6 +53,18 @@ public CollectIteratorAssert<T> withNumRecordsLimit(int limit) {
         return this;
     }
 
+    /**
+     * This method is required for downstream projects e.g. Flink connectors extending this test for
+     * the case when there should be supported Flink versions below 1.20. Could be removed together
+     * with dropping support for Flink 1.19.
+     */
+    @Deprecated
+    public void matchesRecordsFromSource(
+            List<List<T>> recordsBySplitsFromSource,
+            org.apache.flink.streaming.api.CheckpointingMode semantic) {
+        matchesRecordsFromSource(recordsBySplitsFromSource, convertToCheckpointingMode(semantic));
+    }
+
     public void matchesRecordsFromSource(
             List<List<T>> recordsBySplitsFromSource, CheckpointingMode semantic) {
         for (List<T> recordsFromSplit : recordsBySplitsFromSource) {