fix flink-conf is not obtained in local mode

Author: MOBIN-F

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java

@@ -76,7 +76,8 @@ public PipelineExecution.ExecutionInfo run() throws Exception {
             case YARN_APPLICATION:
                 return deployWithApplicationComposer(new YarnApplicationDeploymentExecutor());
             case LOCAL:
-                return deployWithComposer(FlinkPipelineComposer.ofMiniCluster());
+                return deployWithComposer(
+                        FlinkPipelineComposer.ofMiniCluster(flinkConfig, additionalJars));
             case REMOTE:
             case YARN_SESSION:
                 return deployWithComposer(

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java

@@ -18,6 +18,7 @@
 package org.apache.flink.cdc.composer.flink;
 
 import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
 import org.apache.flink.cdc.common.configuration.Configuration;
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.pipeline.PipelineOptions;
@@ -59,32 +60,28 @@ public class FlinkPipelineComposer implements PipelineComposer {
     public static FlinkPipelineComposer ofRemoteCluster(
             org.apache.flink.configuration.Configuration flinkConfig, List<Path> additionalJars) {
         StreamExecutionEnvironment env = new StreamExecutionEnvironment(flinkConfig);
-        additionalJars.forEach(
-                jarPath -> {
-                    try {
-                        FlinkEnvironmentUtils.addJar(
-                                env,
-                                jarPath.makeQualified(jarPath.getFileSystem()).toUri().toURL());
-                    } catch (Exception e) {
-                        throw new RuntimeException(
-                                String.format(
-                                        "Unable to convert JAR path \"%s\" to URL when adding JAR to Flink environment",
-                                        jarPath),
-                                e);
-                    }
-                });
+        addAdditionalJars(env, additionalJars);
         return new FlinkPipelineComposer(env, false);
     }
 
     public static FlinkPipelineComposer ofApplicationCluster(StreamExecutionEnvironment env) {
         return new FlinkPipelineComposer(env, false);
     }
 
+    @VisibleForTesting
     public static FlinkPipelineComposer ofMiniCluster() {
         return new FlinkPipelineComposer(
                 StreamExecutionEnvironment.getExecutionEnvironment(), true);
     }
 
+    public static FlinkPipelineComposer ofMiniCluster(
+            org.apache.flink.configuration.Configuration flinkConfig, List<Path> additionalJars) {
+        StreamExecutionEnvironment localEnvironment =
+                StreamExecutionEnvironment.getExecutionEnvironment(flinkConfig);
+        addAdditionalJars(localEnvironment, additionalJars);
+        return new FlinkPipelineComposer(localEnvironment, true);
+    }
+
     private FlinkPipelineComposer(StreamExecutionEnvironment env, boolean isBlocking) {
         this.env = env;
         this.isBlocking = isBlocking;
@@ -254,4 +251,22 @@ private Optional<URL> getContainingJar(Class<?> clazz) throws Exception {
         }
         return Optional.of(container);
     }
+
+    private static void addAdditionalJars(
+            StreamExecutionEnvironment env, List<Path> additionalJars) {
+        additionalJars.forEach(
+                jarPath -> {
+                    try {
+                        FlinkEnvironmentUtils.addJar(
+                                env,
+                                jarPath.makeQualified(jarPath.getFileSystem()).toUri().toURL());
+                    } catch (Exception e) {
+                        throw new RuntimeException(
+                                String.format(
+                                        "Unable to convert JAR path \"%s\" to URL when adding JAR to Flink environment",
+                                        jarPath),
+                                e);
+                    }
+                });
+    }
 }