[Feature][env]Load seatunnel-env into job env

seatunnel-common/pom.xml

@@ -74,6 +74,12 @@
             <groupId>commons-codec</groupId>
             <artifactId>commons-codec</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.junit-pioneer</groupId>
+            <artifactId>junit-pioneer</artifactId>
+            <version>1.9.1</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

seatunnel-common/src/main/java/org/apache/seatunnel/common/config/Common.java

@@ -143,6 +143,11 @@ public static Path libDir() {
         return Paths.get(getSeaTunnelHome(), "lib");
     }
 
+    /** config Dir */
+    public static Path configDir() {
+        return Paths.get(getSeaTunnelHome(), "config");
+    }
+
     /** return lib jars, which located in 'lib/*' or 'lib/{dir}/*'. */
     public static List<Path> getLibJars() {
         Path libRootDir = Common.libDir();

seatunnel-common/src/main/java/org/apache/seatunnel/common/config/EnvConfigParserUtil.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.common.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class EnvConfigParserUtil {
+
+    public static Map<String, String> parseEnvConfig() {
+        return new HashMap<>(System.getenv());
+    }
+}

seatunnel-common/src/test/java/org/apache/seatunnel/common/config/EnvConfigParserUtilTest.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.common.config;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junitpioneer.jupiter.SetEnvironmentVariable;
+
+import java.util.Map;
+
+public class EnvConfigParserUtilTest {
+
+    @Test
+    @SetEnvironmentVariable(key = "A", value = "B")
+    public void testParseEnvConfig() {
+        Map<String, String> envMap = EnvConfigParserUtil.parseEnvConfig();
+        Assertions.assertEquals("B", envMap.get("A"));
+    }
+}

seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/command/FlinkTaskExecuteCommand.java

@@ -21,7 +21,9 @@
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigUtil;
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueFactory;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.common.Constants;
+import org.apache.seatunnel.common.config.EnvConfigParserUtil;
 import org.apache.seatunnel.core.starter.command.Command;
 import org.apache.seatunnel.core.starter.exception.CommandExecuteException;
 import org.apache.seatunnel.core.starter.flink.args.FlinkCommandArgs;
@@ -32,6 +34,8 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
 
 import static org.apache.seatunnel.core.starter.utils.FileUtils.checkConfigExist;
 
@@ -56,6 +60,13 @@ public void execute() throws CommandExecuteException {
                             ConfigUtil.joinPath("env", "job.name"),
                             ConfigValueFactory.fromAnyRef(flinkCommandArgs.getJobName()));
         }
+        ReadonlyConfig envOptions = ReadonlyConfig.fromConfig(config.getConfig("env"));
+        Map<String, String> envConfigMap = EnvConfigParserUtil.parseEnvConfig();
+        if (envConfigMap != null) {
+            Map<String, Object> mergedConfig = new HashMap<>(envOptions.toMap());
+            mergedConfig.putAll(envConfigMap);
+            config = config.withValue("env", ConfigValueFactory.fromMap(mergedConfig));
+        }
         FlinkExecution seaTunnelTaskExecution = new FlinkExecution(config);
         try {
             seaTunnelTaskExecution.execute();

seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommand.java

@@ -21,7 +21,9 @@
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigUtil;
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueFactory;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.common.Constants;
+import org.apache.seatunnel.common.config.EnvConfigParserUtil;
 import org.apache.seatunnel.core.starter.command.Command;
 import org.apache.seatunnel.core.starter.exception.CommandExecuteException;
 import org.apache.seatunnel.core.starter.spark.args.SparkCommandArgs;
@@ -32,6 +34,8 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
 
 import static org.apache.seatunnel.core.starter.utils.FileUtils.checkConfigExist;
 
@@ -55,6 +59,14 @@ public void execute() throws CommandExecuteException {
                             ConfigUtil.joinPath("env", "job.name"),
                             ConfigValueFactory.fromAnyRef(sparkCommandArgs.getJobName()));
         }
+        ReadonlyConfig envOptions = ReadonlyConfig.fromConfig(config.getConfig("env"));
+        Map<String, String> envConfigMap = EnvConfigParserUtil.parseEnvConfig();
+        if (envConfigMap != null) {
+            Map<String, Object> mergedConfig = new HashMap<>(envOptions.toMap());
+            mergedConfig.putAll(envConfigMap);
+            config = config.withValue("env", ConfigValueFactory.fromMap(mergedConfig));
+        }
+
         try {
             SparkExecution seaTunnelTaskExecution = new SparkExecution(config);
             seaTunnelTaskExecution.execute();

seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java

@@ -41,6 +41,7 @@
 import org.apache.seatunnel.api.transform.SeaTunnelTransform;
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.common.config.Common;
+import org.apache.seatunnel.common.config.EnvConfigParserUtil;
 import org.apache.seatunnel.common.config.TypesafeConfigUtils;
 import org.apache.seatunnel.common.constants.CollectionConstants;
 import org.apache.seatunnel.common.constants.JobMode;
@@ -109,7 +110,7 @@ public class MultipleTableJobConfigParser {
     private final List<URL> commonPluginJars;
     private final Config seaTunnelJobConfig;
 
-    private final ReadonlyConfig envOptions;
+    private ReadonlyConfig envOptions;
 
     private final boolean isStartWithSavePoint;
     private final List<JobPipelineCheckpointData> pipelineCheckpoints;
@@ -314,6 +315,14 @@ private void fillJobConfigAndCommonJars() {
                 || jobConfig.getName().equals(EnvCommonOptions.JOB_NAME.defaultValue())) {
             jobConfig.setName(envOptions.get(EnvCommonOptions.JOB_NAME));
         }
+
+        Map<String, String> envConfigMap = EnvConfigParserUtil.parseEnvConfig();
+        if (envConfigMap != null) {
+            Map<String, Object> mergedConfig = new HashMap<>(envOptions.toMap());
+            mergedConfig.putAll(envConfigMap);
+            envOptions = ReadonlyConfig.fromMap(mergedConfig);
+        }
+
         jobConfig.getEnvOptions().putAll(envOptions.getSourceMap());
         this.commonPluginJars.addAll(
                 new ArrayList<>(