1

liugddx · Jan 23, 2025 · 2bb258f · 2bb258f
1 parent 180b031
commit 2bb258f
Show file tree

Hide file tree

Showing 12 changed files with 30 additions and 54 deletions.
diff --git a/...tunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/enums/PluginType.java b/...tunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/enums/PluginType.java
diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -38,6 +38,7 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.core.starter.enums.EngineType;
 import org.apache.seatunnel.core.starter.exception.TaskExecuteException;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelFactoryDiscovery;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelSinkPluginDiscovery;
@@ -84,7 +85,7 @@ protected List<Optional<? extends Factory>> initializePlugins(
                         (TableSinkFactory)
                                 factoryDiscovery.createPluginInstance(
                                         PluginIdentifier.of(
-                                                ENGINE_TYPE,
+                                                EngineType.SEATUNNEL.getEngine(),
                                                 PluginType.SINK.getType(),
                                                 pluginName));
 
@@ -95,7 +96,7 @@ protected List<Optional<? extends Factory>> initializePlugins(
                                     sinkPluginDiscovery.getPluginJarPaths(
                                             Lists.newArrayList(
                                                     PluginIdentifier.of(
-                                                            ENGINE_TYPE,
+                                                            EngineType.SEATUNNEL.getEngine(),
                                                             PluginType.SINK.getType(),
                                                             sinkConfig.getString(
                                                                     PLUGIN_NAME.key())))));

diff --git a/...rg/apache/seatunnel/core/starter/flink/execution/FlinkAbstractPluginExecuteProcessor.java b/...rg/apache/seatunnel/core/starter/flink/execution/FlinkAbstractPluginExecuteProcessor.java
@@ -36,8 +36,6 @@
 public abstract class FlinkAbstractPluginExecuteProcessor<T>
         implements PluginExecuteProcessor<DataStreamTableInfo, FlinkRuntimeEnvironment> {
 
-    protected static final String ENGINE_TYPE = "seatunnel";
-
     protected static final BiConsumer<ClassLoader, URL> ADD_URL_TO_CLASSLOADER =
             (classLoader, url) -> {
                 if (classLoader.getClass().getName().endsWith("SafetyNetWrapperClassLoader")) {

diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -38,6 +38,7 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.core.starter.enums.EngineType;
 import org.apache.seatunnel.core.starter.exception.TaskExecuteException;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelFactoryDiscovery;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelSinkPluginDiscovery;
@@ -86,7 +87,7 @@ protected List<Optional<? extends Factory>> initializePlugins(
                         (TableSinkFactory)
                                 factoryDiscovery.createPluginInstance(
                                         PluginIdentifier.of(
-                                                ENGINE_TYPE,
+                                                EngineType.SEATUNNEL.getEngine(),
                                                 PluginType.SINK.getType(),
                                                 pluginName));
 
@@ -97,7 +98,7 @@ protected List<Optional<? extends Factory>> initializePlugins(
                                     sinkPluginDiscovery.getPluginJarPaths(
                                             Lists.newArrayList(
                                                     PluginIdentifier.of(
-                                                            ENGINE_TYPE,
+                                                            EngineType.SEATUNNEL.getEngine(),
                                                             PluginType.SINK.getType(),
                                                             sinkConfig.getString(
                                                                     PLUGIN_NAME.key())))));

diff --git a/...c/main/java/org/apache/seatunnel/core/starter/flink/execution/SourceExecuteProcessor.java b/...c/main/java/org/apache/seatunnel/core/starter/flink/execution/SourceExecuteProcessor.java
@@ -30,7 +30,8 @@
 import org.apache.seatunnel.api.table.factory.FactoryUtil;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.core.starter.enums.PluginType;
+import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.core.starter.enums.EngineType;
 import org.apache.seatunnel.core.starter.execution.SourceTableInfo;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelFactoryDiscovery;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelSourcePluginDiscovery;
@@ -58,7 +59,6 @@
 @Slf4j
 @SuppressWarnings("unchecked,rawtypes")
 public class SourceExecuteProcessor extends FlinkAbstractPluginExecuteProcessor<SourceTableInfo> {
-    private static final String PLUGIN_TYPE = PluginType.SOURCE.getType();
 
     public SourceExecuteProcessor(
             List<URL> jarPaths,
@@ -113,7 +113,9 @@ protected List<SourceTableInfo> initializePlugins(
         for (Config sourceConfig : pluginConfigs) {
             PluginIdentifier pluginIdentifier =
                     PluginIdentifier.of(
-                            ENGINE_TYPE, PLUGIN_TYPE, sourceConfig.getString(PLUGIN_NAME.key()));
+                            EngineType.SEATUNNEL.getEngine(),
+                            PluginType.SOURCE.getType(),
+                            sourceConfig.getString(PLUGIN_NAME.key()));
             jars.addAll(
                     sourcePluginDiscovery.getPluginJarPaths(Lists.newArrayList(pluginIdentifier)));
 

diff --git a/...ain/java/org/apache/seatunnel/core/starter/flink/execution/TransformExecuteProcessor.java b/...ain/java/org/apache/seatunnel/core/starter/flink/execution/TransformExecuteProcessor.java
@@ -31,6 +31,7 @@
 import org.apache.seatunnel.api.transform.SeaTunnelMapTransform;
 import org.apache.seatunnel.api.transform.SeaTunnelTransform;
 import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.core.starter.enums.EngineType;
 import org.apache.seatunnel.core.starter.exception.TaskExecuteException;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelFactoryDiscovery;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelTransformPluginDiscovery;
@@ -79,15 +80,18 @@ protected List<TableTransformFactory> initializePlugins(
                 pluginName ->
                         (TableTransformFactory)
                                 factoryDiscovery.createPluginInstance(
-                                        PluginIdentifier.of(ENGINE_TYPE, "transform", pluginName));
+                                        PluginIdentifier.of(
+                                                EngineType.SEATUNNEL.getEngine(),
+                                                PluginType.TRANSFORM.getType(),
+                                                pluginName));
         return pluginConfigs.stream()
                 .map(
                         transformConfig -> {
                             jarPaths.addAll(
                                     transformPluginDiscovery.getPluginJarPaths(
                                             Lists.newArrayList(
                                                     PluginIdentifier.of(
-                                                            ENGINE_TYPE,
+                                                            EngineType.SEATUNNEL.getEngine(),
                                                             PluginType.TRANSFORM.getType(),
                                                             transformConfig.getString(
                                                                     PLUGIN_NAME.key())))));

diff --git a/...l-spark-2-starter/src/main/java/org/apache/seatunnel/core/starter/spark/SparkStarter.java b/...l-spark-2-starter/src/main/java/org/apache/seatunnel/core/starter/spark/SparkStarter.java
@@ -23,9 +23,9 @@
 import org.apache.seatunnel.api.env.EnvCommonOptions;
 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.starter.Starter;
 import org.apache.seatunnel.core.starter.enums.EngineType;
-import org.apache.seatunnel.core.starter.enums.PluginType;
 import org.apache.seatunnel.core.starter.spark.args.SparkCommandArgs;
 import org.apache.seatunnel.core.starter.utils.CommandLineUtils;
 import org.apache.seatunnel.core.starter.utils.CompressionUtils;

diff --git a/...rk-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/SparkStarter.java b/...rk-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/SparkStarter.java
@@ -23,9 +23,9 @@
 import org.apache.seatunnel.api.env.EnvCommonOptions;
 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.starter.Starter;
 import org.apache.seatunnel.core.starter.enums.EngineType;
-import org.apache.seatunnel.core.starter.enums.PluginType;
 import org.apache.seatunnel.core.starter.spark.args.SparkCommandArgs;
 import org.apache.seatunnel.core.starter.utils.CommandLineUtils;
 import org.apache.seatunnel.core.starter.utils.CompressionUtils;

diff --git a/...c/main/java/org/apache/seatunnel/core/starter/spark/execution/SourceExecuteProcessor.java b/...c/main/java/org/apache/seatunnel/core/starter/spark/execution/SourceExecuteProcessor.java
@@ -30,7 +30,9 @@
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.factory.FactoryUtil;
 import org.apache.seatunnel.common.Constants;
+import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.common.utils.SerializationUtils;
+import org.apache.seatunnel.core.starter.enums.EngineType;
 import org.apache.seatunnel.core.starter.execution.SourceTableInfo;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelSourcePluginDiscovery;
 import org.apache.seatunnel.translation.spark.execution.DatasetTableInfo;
@@ -56,7 +58,6 @@
 
 @SuppressWarnings("rawtypes")
 public class SourceExecuteProcessor extends SparkAbstractPluginExecuteProcessor<SourceTableInfo> {
-    private static final String PLUGIN_TYPE = "source";
     private Map envOption = new HashMap<String, String>();
 
     public SourceExecuteProcessor(
@@ -124,7 +125,9 @@ protected List<SourceTableInfo> initializePlugins(List<? extends Config> pluginC
         for (Config sourceConfig : pluginConfigs) {
             PluginIdentifier pluginIdentifier =
                     PluginIdentifier.of(
-                            ENGINE_TYPE, PLUGIN_TYPE, sourceConfig.getString(PLUGIN_NAME.key()));
+                            EngineType.SEATUNNEL.getEngine(),
+                            PluginType.SOURCE.getType(),
+                            sourceConfig.getString(PLUGIN_NAME.key()));
             jars.addAll(
                     sourcePluginDiscovery.getPluginJarPaths(Lists.newArrayList(pluginIdentifier)));
             Tuple2<SeaTunnelSource<Object, SourceSplit, Serializable>, List<CatalogTable>> source =

diff --git a/...rg/apache/seatunnel/core/starter/spark/execution/SparkAbstractPluginExecuteProcessor.java b/...rg/apache/seatunnel/core/starter/spark/execution/SparkAbstractPluginExecuteProcessor.java
@@ -48,7 +48,6 @@ public abstract class SparkAbstractPluginExecuteProcessor<T>
     protected final List<? extends Config> pluginConfigs;
     protected final JobContext jobContext;
     protected final List<T> plugins;
-    protected static final String ENGINE_TYPE = "seatunnel";
     protected final ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
 
     protected SparkAbstractPluginExecuteProcessor(

diff --git a/.../main/java/org/apache/seatunnel/core/starter/spark/execution/SparkRuntimeEnvironment.java b/.../main/java/org/apache/seatunnel/core/starter/spark/execution/SparkRuntimeEnvironment.java
@@ -22,7 +22,7 @@
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.common.config.CheckResult;
 import org.apache.seatunnel.common.constants.JobMode;
-import org.apache.seatunnel.core.starter.enums.PluginType;
+import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.core.starter.execution.RuntimeEnvironment;
 
 import org.apache.spark.SparkConf;

diff --git a/...ain/java/org/apache/seatunnel/core/starter/spark/execution/TransformExecuteProcessor.java b/...ain/java/org/apache/seatunnel/core/starter/spark/execution/TransformExecuteProcessor.java
@@ -31,6 +31,8 @@
 import org.apache.seatunnel.api.transform.SeaTunnelFlatMapTransform;
 import org.apache.seatunnel.api.transform.SeaTunnelMapTransform;
 import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.core.starter.enums.EngineType;
 import org.apache.seatunnel.core.starter.exception.TaskExecuteException;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelTransformPluginDiscovery;
 import org.apache.seatunnel.translation.spark.execution.DatasetTableInfo;
@@ -85,8 +87,9 @@ protected List<TableTransformFactory> initializePlugins(List<? extends Config> p
                                             transformPluginDiscovery.getPluginJarPaths(
                                                     Lists.newArrayList(
                                                             PluginIdentifier.of(
-                                                                    ENGINE_TYPE,
-                                                                    "transform",
+                                                                    EngineType.SEATUNNEL
+                                                                            .getEngine(),
+                                                                    PluginType.TRANSFORM.getType(),
                                                                     transformConfig.getString(
                                                                             PLUGIN_NAME.key())))));
                                     return discoverOptionalFactory(