Add Flink2 support for pipeline source connector.

Author: lvyanquan

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java

@@ -66,6 +66,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
@@ -161,7 +162,12 @@ private int getParallelism() {
     protected String flinkVersion = getFlinkVersion();
 
     public static String getFlinkVersion() {
-        return "1.20.3";
+        String flinkVersion = System.getProperty("specifiedFlinkVersion");
+        if (Objects.isNull(flinkVersion)) {
+            throw new IllegalArgumentException(
+                    "No Flink version specified to run this test. Please use -DspecifiedFlinkVersion to pass one.");
+        }
+        return flinkVersion;
     }
 
     protected List<String> copyJarToFlinkLib() {
@@ -398,6 +404,17 @@ public void waitUntilJobState(Duration timeout, JobStatus expectedStatus) {
     }
 
     protected String getFlinkDockerImageTag() {
+        String customImage = System.getProperty("flink.docker.image");
+        if (customImage != null && !customImage.isEmpty()) {
+            return customImage;
+        }
+        String registry = System.getProperty("flink.docker.registry", "");
+        if (!registry.isEmpty()) {
+            if (System.getProperty("java.specification.version").equals("17")) {
+                return String.format("%s/flink:%s-scala_2.12-java17", registry, flinkVersion);
+            }
+            return String.format("%s/flink:%s-scala_2.12-java11", registry, flinkVersion);
+        }
         if (System.getProperty("java.specification.version").equals("17")) {
             return String.format("flink:%s-scala_2.12-java17", flinkVersion);
         }