[FLINK-29479][python] Supports whether uses system env for python.

This closes apache#21110.
BugsEverywhere · Oct 25, 2022 · 8e16cc8 · 8e16cc8
1 parent 0e61285
commit 8e16cc8
Show file tree

Hide file tree

Showing 6 changed files with 36 additions and 2 deletions.
diff --git a/docs/layouts/shortcodes/generated/python_configuration.html b/docs/layouts/shortcodes/generated/python_configuration.html
@@ -110,5 +110,11 @@
             <td>Integer</td>
             <td>The maximum number of states cached in a Python UDF worker. Note that this is an experimental flag and might not be available in future releases.</td>
         </tr>
+        <tr>
+            <td><h5>python.systemenv.enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Specify whether to load System Environment when starting Python worker.</td>
+        </tr>
     </tbody>
 </table>
diff --git a/flink-python/src/main/java/org/apache/flink/python/PythonOptions.java b/flink-python/src/main/java/org/apache/flink/python/PythonOptions.java
@@ -83,6 +83,14 @@ public class PythonOptions {
                                     + "The interval between each profiling is determined by the config options "
                                     + "python.fn-execution.bundle.size and python.fn-execution.bundle.time.");
 
+    /** The configuration to enable or disable system env for Python execution. */
+    public static final ConfigOption<Boolean> PYTHON_SYSTEMENV_ENABLED =
+            ConfigOptions.key("python.systemenv.enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Specify whether to load System Environment when starting Python worker.");
+
     /** The configuration to enable or disable python operator chaining. */
     public static final ConfigOption<Boolean> PYTHON_OPERATOR_CHAINING_ENABLED =
             ConfigOptions.key("python.operator-chaining.enabled")

diff --git a/.../java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java b/.../java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
@@ -40,6 +40,7 @@
 import static org.apache.flink.python.PythonOptions.MAX_BUNDLE_SIZE;
 import static org.apache.flink.python.PythonOptions.MAX_BUNDLE_TIME_MILLS;
 import static org.apache.flink.python.PythonOptions.PYTHON_METRIC_ENABLED;
+import static org.apache.flink.python.PythonOptions.PYTHON_SYSTEMENV_ENABLED;
 import static org.apache.flink.streaming.api.utils.ClassLeakCleaner.cleanUpLeakingClasses;
 import static org.apache.flink.streaming.api.utils.PythonOperatorUtils.inBatchExecutionMode;
 
@@ -51,6 +52,8 @@ public abstract class AbstractPythonFunctionOperator<OUT> extends AbstractStream
 
     protected final Configuration config;
 
+    protected transient boolean systemEnvEnabled;
+
     /** Max number of elements to include in a bundle. */
     protected transient int maxBundleSize;
 
@@ -77,6 +80,7 @@ public AbstractPythonFunctionOperator(Configuration config) {
     @Override
     public void open() throws Exception {
         try {
+            this.systemEnvEnabled = config.get(PYTHON_SYSTEMENV_ENABLED);
             this.maxBundleSize = config.get(MAX_BUNDLE_SIZE);
             if (this.maxBundleSize <= 0) {
                 this.maxBundleSize = MAX_BUNDLE_SIZE.defaultValue();

diff --git a/...flink/streaming/api/operators/python/embedded/AbstractEmbeddedPythonFunctionOperator.java b/...flink/streaming/api/operators/python/embedded/AbstractEmbeddedPythonFunctionOperator.java
@@ -141,7 +141,7 @@ protected EmbeddedPythonEnvironmentManager createPythonEnvironmentManager() {
         return new EmbeddedPythonEnvironmentManager(
                 dependencyInfo,
                 getContainingTask().getEnvironment().getTaskManagerInfo().getTmpDirectories(),
-                new HashMap<>(System.getenv()),
+                systemEnvEnabled ? new HashMap<>(System.getenv()) : new HashMap<>(),
                 getRuntimeContext().getJobId());
     }
 

diff --git a/.../flink/streaming/api/operators/python/process/AbstractExternalPythonFunctionOperator.java b/.../flink/streaming/api/operators/python/process/AbstractExternalPythonFunctionOperator.java
@@ -127,7 +127,7 @@ protected ProcessPythonEnvironmentManager createPythonEnvironmentManager() {
             return new ProcessPythonEnvironmentManager(
                     dependencyInfo,
                     getContainingTask().getEnvironment().getTaskManagerInfo().getTmpDirectories(),
-                    new HashMap<>(System.getenv()),
+                    systemEnvEnabled ? new HashMap<>(System.getenv()) : new HashMap<>(),
                     getRuntimeContext().getJobId());
         } else {
             throw new UnsupportedOperationException(

diff --git a/flink-python/src/test/java/org/apache/flink/python/PythonOptionsTest.java b/flink-python/src/test/java/org/apache/flink/python/PythonOptionsTest.java
@@ -172,4 +172,20 @@ void testPythonClientExecutable() {
                 configuration.get(PythonOptions.PYTHON_CLIENT_EXECUTABLE);
         assertThat(actualPythonClientExecutable).isEqualTo(expectedPythonClientExecutable);
     }
+
+    @Test
+    void testPythonSystemEnvEnabled() {
+        final Configuration configuration = new Configuration();
+        final boolean isSystemEnvEnabled =
+                configuration.getBoolean(PythonOptions.PYTHON_SYSTEMENV_ENABLED);
+        assertThat(isSystemEnvEnabled)
+                .isEqualTo(PythonOptions.PYTHON_SYSTEMENV_ENABLED.defaultValue());
+
+        final boolean expectedIsSystemEnvEnabled = false;
+        configuration.setBoolean(PythonOptions.PYTHON_SYSTEMENV_ENABLED, false);
+
+        final boolean actualIsSystemEnvEnabled =
+                configuration.getBoolean(PythonOptions.PYTHON_SYSTEMENV_ENABLED);
+        assertThat(actualIsSystemEnvEnabled).isEqualTo(expectedIsSystemEnvEnabled);
+    }
 }