[improve]Transform the End-To-End(E2E) tasks on the assembly line (ap…

…ache#466)

.licenserc.yaml

@@ -12,5 +12,6 @@ header:
     - '.github/PULL_REQUEST_TEMPLATE.md'
     - '.licenserc.yaml'
     - 'custom_env.sh.tpl'
+    - 'flink-doris-connector/src/test/resources/container/'
 
   comment: on-failure

flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/CdcTools.java

@@ -20,6 +20,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.java.utils.MultipleParameterTool;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.StringUtils;
@@ -36,11 +37,14 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /** cdc sync tools. */
 public class CdcTools {
     private static final List<String> EMPTY_KEYS =
             Collections.singletonList(DatabaseSyncConfig.PASSWORD);
+    private static StreamExecutionEnvironment flinkEnvironmentForTesting;
+    private static JobClient jobClient;
 
     public static void main(String[] args) throws Exception {
         System.out.println("Input args: " + Arrays.asList(args) + ".\n");
@@ -146,7 +150,10 @@ private static void syncDatabase(
                 new DorisTableConfig(getConfigMap(params, DatabaseSyncConfig.TABLE_CONF));
         Configuration sinkConfig = Configuration.fromMap(sinkMap);
 
-        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        StreamExecutionEnvironment env =
+                Objects.nonNull(flinkEnvironmentForTesting)
+                        ? flinkEnvironmentForTesting
+                        : StreamExecutionEnvironment.getExecutionEnvironment();
         databaseSync
                 .setEnv(env)
                 .setDatabase(database)
@@ -174,7 +181,23 @@ private static void syncDatabase(
                             config.getString(
                                     DatabaseSyncConfig.DATABASE_NAME, DatabaseSyncConfig.DB));
         }
-        env.execute(jobName);
+        if (Objects.nonNull(flinkEnvironmentForTesting)) {
+            jobClient = env.executeAsync();
+        } else {
+            env.execute(jobName);
+        }
+    }
+
+    @VisibleForTesting
+    public static JobClient getJobClient() {
+        return jobClient;
+    }
+
+    // Only for testing, please do not use it in actual environment
+    @VisibleForTesting
+    public static void setStreamExecutionEnvironmentForTesting(
+            StreamExecutionEnvironment environment) {
+        flinkEnvironmentForTesting = environment;
     }
 
     @VisibleForTesting