[Fix-6497][Server] Fix the missing code about the feature of supporting ...

[Fix-6497][Server] Fix the missing code about the feature of supporting  multiple environments. (#6509) (#6510)
Co-authored-by: Hua Jiang <jianghuachinacom@163.com>

dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/task/request/TaskRequest.java

@@ -148,6 +148,11 @@ public class TaskRequest {
      */
     private String envFile;
 
+    /**
+     * environmentConfig
+     */
+    private String environmentConfig;
+
     /**
      * definedParams
      */
@@ -412,6 +417,14 @@ public class TaskRequest {
         this.envFile = envFile;
     }
 
+    public String getEnvironmentConfig() {
+        return environmentConfig;
+    }
+
+    public void setEnvironmentConfig(String config) {
+        this.environmentConfig = config;
+    }
+
     public Map<String, String> getDefinedParams() {
         return definedParams;
     }

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/ShellCommandExecutor.java

@@ -21,13 +21,13 @@ import org.apache.dolphinscheduler.plugin.task.util.OSUtils;
 import org.apache.dolphinscheduler.spi.task.request.TaskRequest;
 
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.StringUtils;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
-import java.util.List;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.function.Consumer;
 
@@ -94,18 +94,25 @@ public class ShellCommandExecutor extends AbstractCommandExecutor {
             if (OSUtils.isWindows()) {
                 sb.append("@echo off\n");
                 sb.append("cd /d %~dp0\n");
-                if (taskRequest.getEnvFile() != null) {
-                    sb.append("call ").append(taskRequest.getEnvFile()).append("\n");
+                if (StringUtils.isNotBlank(taskRequest.getEnvironmentConfig())) {
+                    sb.append(taskRequest.getEnvironmentConfig()).append("\n");
+                } else {
+                    if (taskRequest.getEnvFile() != null) {
+                        sb.append("call ").append(taskRequest.getEnvFile()).append("\n");
+                    }
                 }
             } else {
                 sb.append("#!/bin/sh\n");
                 sb.append("BASEDIR=$(cd `dirname $0`; pwd)\n");
                 sb.append("cd $BASEDIR\n");
-                if (taskRequest.getEnvFile() != null) {
-                    sb.append("source ").append(taskRequest.getEnvFile()).append("\n");
+                if (StringUtils.isNotBlank(taskRequest.getEnvironmentConfig())) {
+                    sb.append(taskRequest.getEnvironmentConfig()).append("\n");
+                } else {
+                    if (taskRequest.getEnvFile() != null) {
+                        sb.append("source ").append(taskRequest.getEnvFile()).append("\n");
+                    }
                 }
             }
-
             sb.append(execCommand);
             logger.info("command : {}", sb);
 

dolphinscheduler-task-plugin/dolphinscheduler-task-python/src/main/java/org/apache/dolphinscheduler/plugin/task/python/PythonCommandExecutor.java

@@ -17,6 +17,7 @@
 
 package org.apache.dolphinscheduler.plugin.task.python;
 
+import java.util.Arrays;
 import org.apache.dolphinscheduler.plugin.task.api.AbstractCommandExecutor;
 import org.apache.dolphinscheduler.spi.task.request.TaskRequest;
 import org.apache.dolphinscheduler.spi.utils.StringUtils;
@@ -153,6 +154,11 @@ public class PythonCommandExecutor extends AbstractCommandExecutor {
     @Override
     protected String commandInterpreter() {
         String pythonHome = getPythonHome(taskRequest.getEnvFile());
+
+        if (StringUtils.isNotBlank(taskRequest.getEnvironmentConfig())) {
+            pythonHome = getPythonHomeFromEnvironmentConfig(taskRequest.getEnvironmentConfig());
+        }
+
         return getPythonCommand(pythonHome);
     }
 
@@ -176,4 +182,24 @@ public class PythonCommandExecutor extends AbstractCommandExecutor {
         return Paths.get(pythonHome, "/bin/python").toString();
     }
 
+    /**
+     * get python home from the environment config
+     *
+     * @param environmentConfig env config
+     * @return python home
+     */
+    public static String getPythonHomeFromEnvironmentConfig(String environmentConfig) {
+        String[] lines = environmentConfig.split("\n");
+
+        String pythonHomeConfig = Arrays.stream(lines).filter(line -> line.contains(PythonConstants.PYTHON_HOME)).findFirst().get();
+
+        if (StringUtils.isEmpty(pythonHomeConfig)) {
+            return null;
+        }
+        String[] arrs = pythonHomeConfig.split(PythonConstants.EQUAL_SIGN);
+        if (arrs.length == 2) {
+            return arrs[1];
+        }
+        return null;
+    }
 }