[API-DRAFT] [MERGE] Fix obvious bugs that don't work properly before merge

config/flink.batch.conf.template

@@ -26,12 +26,11 @@ env {
 
 source {
   # This is a example input plugin **only for test and demonstrate the feature input plugin**
-  FileSource {
-    path = "hdfs://localhost:9000/output/text"
-    format.type = "text"
-    schema = "string"
-    result_table_name = "test"
-  }
+    FakeSource {
+      result_table_name = "test"
+      field_name = "name,age"
+    }
+
 
   # If you would like to get more information about how to configure seatunnel and see full list of input plugins,
   # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake

config/spark.streaming.conf.template

@@ -31,7 +31,7 @@ env {
 
 source {
   # This is a example input plugin **only for test and demonstrate the feature input plugin**
-  fakeStream {
+  FakeStream {
     content = ["Hello World, SeaTunnel"]
   }
 

plugin-mapping.properties

@@ -45,6 +45,7 @@ flink.sink.Kafka = seatunnel-connector-flink-kafka
 
 spark.source.ElasticSearch = seatunnel-connector-spark-elasticsearch
 spark.source.Fake = seatunnel-connector-spark-fake
+spark.source.FakeStream = seatunnel-connector-spark-fake
 spark.source.FeishuSheet = seatunnel-connector-spark-feishu
 spark.source.File = seatunnel-connector-spark-file
 spark.source.Hbase = seatunnel-connector-spark-hbase

pom.xml

@@ -82,15 +82,15 @@
         <module>seatunnel-core</module>
         <module>seatunnel-transforms</module>
         <module>seatunnel-connectors</module>
-        <module>seatunnel-dist</module>
+        <module>seatunnel-connectors-v2</module>
+        <module>seatunnel-connectors-v2-dist</module>
         <module>seatunnel-examples</module>
         <module>seatunnel-e2e</module>
         <module>seatunnel-api</module>
         <module>seatunnel-translation</module>
         <module>seatunnel-plugin-discovery</module>
         <module>seatunnel-formats</module>
-        <module>seatunnel-connectors-v2</module>
-        <module>seatunnel-connectors-v2-dist</module>
+        <module>seatunnel-dist</module>
     </modules>
 
     <properties>

seatunnel-core/seatunnel-core-base/src/main/java/org/apache/seatunnel/core/base/config/AbstractExecutionContext.java

@@ -80,16 +80,13 @@ public JobMode getJobMode() {
 
     @SuppressWarnings("checkstyle:Indentation")
     protected List<PluginIdentifier> getPluginIdentifiers(PluginType... pluginTypes) {
-        return Arrays.stream(pluginTypes).flatMap(new Function<PluginType, Stream<PluginIdentifier>>() {
-            @Override
-            public Stream<PluginIdentifier> apply(PluginType pluginType) {
-                List<? extends Config> configList = config.getConfigList(pluginType.getType());
-                return configList.stream()
-                    .map(pluginConfig -> PluginIdentifier
-                        .of(engine.getEngine(),
-                            pluginType.getType(),
-                            pluginConfig.getString("plugin_name")));
-            }
+        return Arrays.stream(pluginTypes).flatMap((Function<PluginType, Stream<PluginIdentifier>>) pluginType -> {
+            List<? extends Config> configList = config.getConfigList(pluginType.getType());
+            return configList.stream()
+                .map(pluginConfig -> PluginIdentifier
+                    .of(engine.getEngine(),
+                        pluginType.getType(),
+                        pluginConfig.getString("plugin_name")));
         }).collect(Collectors.toList());
     }
 }

seatunnel-core/seatunnel-core-spark/src/main/java/org/apache/seatunnel/core/spark/SparkStarter.java

@@ -22,12 +22,15 @@
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.common.config.Common;
 import org.apache.seatunnel.common.config.DeployMode;
+import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.core.base.Starter;
 import org.apache.seatunnel.core.base.config.ConfigBuilder;
 import org.apache.seatunnel.core.base.config.EngineType;
 import org.apache.seatunnel.core.base.utils.CompressionUtils;
 import org.apache.seatunnel.core.spark.args.SparkCommandArgs;
-import org.apache.seatunnel.core.spark.config.SparkExecutionContext;
+import org.apache.seatunnel.plugin.discovery.PluginIdentifier;
+import org.apache.seatunnel.plugin.discovery.spark.SparkSinkPluginDiscovery;
+import org.apache.seatunnel.plugin.discovery.spark.SparkSourcePluginDiscovery;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
@@ -40,6 +43,7 @@
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -50,6 +54,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -220,8 +225,12 @@ private List<Path> getConnectorJarDependencies() {
             return Collections.emptyList();
         }
         Config config = new ConfigBuilder(Paths.get(commandArgs.getConfigFile())).getConfig();
-        SparkExecutionContext sparkExecutionContext = new SparkExecutionContext(config, EngineType.SPARK);
-        return sparkExecutionContext.getPluginJars().stream().map(url -> new File(url.getPath()).toPath()).collect(Collectors.toList());
+        List<URL> pluginJars = new ArrayList<>();
+        SparkSourcePluginDiscovery sparkSourcePluginDiscovery = new SparkSourcePluginDiscovery();
+        SparkSinkPluginDiscovery sparkSinkPluginDiscovery = new SparkSinkPluginDiscovery();
+        pluginJars.addAll(sparkSourcePluginDiscovery.getPluginJarPaths(getPluginIdentifiers(config, PluginType.SOURCE)));
+        pluginJars.addAll(sparkSinkPluginDiscovery.getPluginJarPaths(getPluginIdentifiers(config, PluginType.SINK)));
+        return pluginJars.stream().map(url -> new File(url.getPath()).toPath()).collect(Collectors.toList());
     }
 
     /**
@@ -313,6 +322,18 @@ protected void appendAppJar(List<String> commands) {
         commands.add(Common.appLibDir().resolve("seatunnel-core-spark.jar").toString());
     }
 
+    @SuppressWarnings("checkstyle:Indentation")
+    private List<PluginIdentifier> getPluginIdentifiers(Config config, PluginType... pluginTypes) {
+        return Arrays.stream(pluginTypes).flatMap((Function<PluginType, Stream<PluginIdentifier>>) pluginType -> {
+            List<? extends Config> configList = config.getConfigList(pluginType.getType());
+            return configList.stream()
+                    .map(pluginConfig -> PluginIdentifier
+                            .of(EngineType.SPARK.getEngine(),
+                                    pluginType.getType(),
+                                    pluginConfig.getString("plugin_name")));
+        }).collect(Collectors.toList());
+    }
+
     /**
      * a Starter for building spark-submit commands with client mode options
      */

seatunnel-dist/src/main/assembly/assembly-bin.xml

@@ -146,6 +146,7 @@
             </includes>
             <excludes>
                 <exclude>%regex[.*((javadoc)|(sources))\.jar]</exclude>
+                <exclude>connector-common*.jar</exclude>
             </excludes>
             <outputDirectory>/connectors/seatunnel</outputDirectory>
         </fileSet>