This is an automated email from the ASF dual-hosted git repository. dianfu pushed a commit to branch release-2.2 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 514f9ba25f0ec50b3ab1e84cd9c38aa15bb4c3b8 Author: Martijn Visser <[email protected]> AuthorDate: Fri Sep 18 04:35:14 2026 +0200 [FLINK-39572][python] Do not log the Beam environment payload (#29225) The copy of Beam's DefaultJobBundleFactory maintained in flink-python logged the whole RunnerApi.Environment when closing, expiring or failing to clean up an environment. For process environments the payload carries every environment variable of the Python worker, which is inherited from the TaskManager, so the TaskManager log contained all of them. Only the environment's urn is logged now. --- .../pyflink/table/tests/test_dependency.py | 18 ++- .../control/DefaultJobBundleFactory.java | 6 +- .../env/AbstractPythonEnvironmentManager.java | 31 +++++ .../process/ProcessPythonEnvironmentManager.java | 4 +- .../control/DefaultJobBundleFactoryTest.java | 146 +++++++++++++++++++++ .../ProcessPythonEnvironmentManagerTest.java | 39 ++++++ 6 files changed, 237 insertions(+), 7 deletions(-) diff --git a/flink-python/pyflink/table/tests/test_dependency.py b/flink-python/pyflink/table/tests/test_dependency.py index 126311d0472..8b3b9c9157c 100644 --- a/flink-python/pyflink/table/tests/test_dependency.py +++ b/flink-python/pyflink/table/tests/test_dependency.py @@ -231,18 +231,30 @@ class StreamDependencyTests(DependencyTests, PyFlinkTestCase): udf(check_pyflink_gateway_disabled, DataTypes.BIGINT(), DataTypes.BIGINT())) + def check_inherited_environment(i): + # set by the test setup before the gateway JVM starts; the worker only sees it + # because it inherits the TaskManager's environment + import os + assert os.environ["FLINK_TESTING"] == "1" + return i + + self.t_env.create_temporary_system_function( + "check_inherited_environment", + udf(check_inherited_environment, DataTypes.BIGINT(), DataTypes.BIGINT())) + sink_table_ddl = """ - CREATE TABLE Results(a BIGINT, b BIGINT) WITH ('connector'='test-sink') + CREATE TABLE Results(a BIGINT, b BIGINT, c BIGINT) WITH ('connector'='test-sink') """ self.t_env.execute_sql(sink_table_ddl) t = self.t_env.from_elements([(1, 2), (2, 5), (3, 1)], ['a', 'b']) t.select( expr.call('check_python_exec', t.a), - expr.call('check_pyflink_gateway_disabled', t.a)) \ + expr.call('check_pyflink_gateway_disabled', t.a), + expr.call('check_inherited_environment', t.a)) \ .execute_insert("Results").wait() actual = source_sink_utils.results() - self.assert_equals(actual, ["+I[1, 1]", "+I[2, 2]", "+I[3, 3]"]) + self.assert_equals(actual, ["+I[1, 1, 1]", "+I[2, 2, 2]", "+I[3, 3, 3]"]) if __name__ == "__main__": diff --git a/flink-python/src/main/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactory.java b/flink-python/src/main/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactory.java index 25523f02944..0f81c952da8 100644 --- a/flink-python/src/main/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactory.java +++ b/flink-python/src/main/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactory.java @@ -226,7 +226,7 @@ public class DefaultJobBundleFactory implements JobBundleFactory { if (refCount > 0) { LOG.warn( "Expiring environment {} with {} remaining bundle references. Taking note to clean it up during shutdown if the references are not removed by then.", - notification.getKey(), + notification.getKey().getUrn(), refCount); evictedActiveClients.add(client); } @@ -697,7 +697,7 @@ public class DefaultJobBundleFactory implements JobBundleFactory { // the next one will be added via Throwable#addSuppressed. closed = true; } catch (Exception e) { - LOG.warn("Error cleaning up servers {}", environment.getEnvironment(), e); + LOG.warn("Error cleaning up servers {}", environment.getEnvironment().getUrn(), e); } // TODO: Wait for executor shutdown? } @@ -711,7 +711,7 @@ public class DefaultJobBundleFactory implements JobBundleFactory { Preconditions.checkState(refCount >= 0, "Reference count must not be negative."); if (refCount == 0) { // Close environment after it was removed from cache and all bundles finished. - LOG.info("Closing environment {}", environment.getEnvironment()); + LOG.info("Closing environment {}", environment.getEnvironment().getUrn()); close(); } return refCount; diff --git a/flink-python/src/main/java/org/apache/flink/python/env/AbstractPythonEnvironmentManager.java b/flink-python/src/main/java/org/apache/flink/python/env/AbstractPythonEnvironmentManager.java index f895f492e8b..8e482d43250 100644 --- a/flink-python/src/main/java/org/apache/flink/python/env/AbstractPythonEnvironmentManager.java +++ b/flink-python/src/main/java/org/apache/flink/python/env/AbstractPythonEnvironmentManager.java @@ -98,6 +98,21 @@ public abstract class AbstractPythonEnvironmentManager implements PythonEnvironm @VisibleForTesting public static final String PYFLINK_GATEWAY_DISABLED = "PYFLINK_GATEWAY_DISABLED"; + /** The environment variables this manager sets for the Python worker. */ + @VisibleForTesting + static final List<String> MANAGED_ENVIRONMENT_VARIABLES = + Collections.unmodifiableList( + Arrays.asList( + "PYTHONPATH", + "PATH", + "python", + "BOOT_LOG_DIR", + PYFLINK_GATEWAY_DISABLED, + PYTHON_REQUIREMENTS_FILE, + PYTHON_REQUIREMENTS_CACHE, + PYTHON_REQUIREMENTS_INSTALL_DIR, + PYTHON_WORKING_DIR)); + public AbstractPythonEnvironmentManager( PythonDependencyInfo dependencyInfo, String[] tmpDirectories, @@ -162,6 +177,22 @@ public abstract class AbstractPythonEnvironmentManager implements PythonEnvironm return resource.env; } + /** + * Returns the environment variables to pass on to the Python worker process explicitly, i.e. + * the ones set by this manager ({@link #MANAGED_ENVIRONMENT_VARIABLES}). Everything else the + * worker needs it inherits from the TaskManager process. + */ + public Map<String, String> getManagedEnvironmentVariables() { + final Map<String, String> env = new HashMap<>(); + for (String name : MANAGED_ENVIRONMENT_VARIABLES) { + final String value = resource.env.get(name); + if (value != null) { + env.put(name, value); + } + } + return env; + } + /** * Constructs the environment variables which is used to launch the python UDF worker. * diff --git a/flink-python/src/main/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManager.java b/flink-python/src/main/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManager.java index b316e065b84..a3414fbac33 100644 --- a/flink-python/src/main/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManager.java +++ b/flink-python/src/main/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManager.java @@ -59,7 +59,9 @@ public final class ProcessPythonEnvironmentManager extends AbstractPythonEnviron PythonEnvironmentManagerUtils.getPythonUdfRunnerScript( dependencyInfo.getPythonExec(), env); - return new ProcessPythonEnvironment(runnerScript, env); + // only what this manager sets; the worker inherits the rest of the TaskManager's + // environment + return new ProcessPythonEnvironment(runnerScript, getManagedEnvironmentVariables()); } /** diff --git a/flink-python/src/test/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactoryTest.java b/flink-python/src/test/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactoryTest.java new file mode 100644 index 00000000000..9b63c5afde3 --- /dev/null +++ b/flink-python/src/test/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactoryTest.java @@ -0,0 +1,146 @@ +/* + * 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.beam.runners.fnexecution.control; + +import org.apache.flink.testutils.logging.LoggerAuditingExtension; + +import org.apache.beam.model.fnexecution.v1.BeamFnApi.InstructionRequest; +import org.apache.beam.model.fnexecution.v1.BeamFnApi.InstructionResponse; +import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleDescriptor; +import org.apache.beam.model.pipeline.v1.RunnerApi.Coder; +import org.apache.beam.model.pipeline.v1.RunnerApi.Components; +import org.apache.beam.model.pipeline.v1.RunnerApi.Environment; +import org.apache.beam.model.pipeline.v1.RunnerApi.ExecutableStagePayload; +import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec; +import org.apache.beam.model.pipeline.v1.RunnerApi.PCollection; +import org.apache.beam.model.pipeline.v1.RunnerApi.WindowingStrategy; +import org.apache.beam.runners.core.construction.Environments; +import org.apache.beam.runners.core.construction.ModelCoders; +import org.apache.beam.runners.core.construction.graph.ExecutableStage; +import org.apache.beam.runners.fnexecution.environment.EnvironmentFactory; +import org.apache.beam.runners.fnexecution.environment.RemoteEnvironment; +import org.apache.beam.runners.fnexecution.provisioning.JobInfo; +import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.Struct; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.event.Level; + +import java.util.Collections; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for the copy of {@link DefaultJobBundleFactory} maintained in flink-python. */ +class DefaultJobBundleFactoryTest { + + private static final String SECRET_VALUE = "canary-value-that-must-not-be-logged"; + + @RegisterExtension + private final LoggerAuditingExtension logging = + new LoggerAuditingExtension(DefaultJobBundleFactory.class, Level.INFO); + + @Test + void closingAnEnvironmentDoesNotLogItsPayload() throws Exception { + final Environment environment = + Environments.createProcessEnvironment( + "", + "", + "pyflink-udf-runner.sh", + Collections.singletonMap("MY_SECRET", SECRET_VALUE)); + final EnvironmentFactory environmentFactory = + (env, workerId) -> + RemoteEnvironment.forHandler(env, new AcknowledgingRequestHandler()); + final EnvironmentFactory.Provider environmentFactoryProvider = + (controlServer, + loggingServer, + retrievalServer, + provisioningServer, + clientPool, + idGenerator) -> environmentFactory; + + try (DefaultJobBundleFactory bundleFactory = + new DefaultJobBundleFactory( + JobInfo.create("testJob", "testJob", "token", Struct.getDefaultInstance()), + Collections.singletonMap( + environment.getUrn(), environmentFactoryProvider))) { + // the stage holds a reference on the environment; releasing both closes it + bundleFactory.forStage(getExecutableStage(environment)).close(); + } + + assertThat(logging.getMessages()) + .anySatisfy( + message -> + assertThat(message) + .contains("Closing environment") + .contains(environment.getUrn())) + .noneSatisfy(message -> assertThat(message).contains(SECRET_VALUE)); + } + + /** Answers every instruction immediately, standing in for an SDK harness. */ + private static final class AcknowledgingRequestHandler implements InstructionRequestHandler { + + @Override + public void registerProcessBundleDescriptor(ProcessBundleDescriptor descriptor) {} + + @Override + public CompletionStage<InstructionResponse> handle(InstructionRequest request) { + return CompletableFuture.completedFuture( + InstructionResponse.newBuilder() + .setInstructionId(request.getInstructionId()) + .build()); + } + + @Override + public void close() {} + } + + private static ExecutableStage getExecutableStage(Environment environment) { + return ExecutableStage.fromPayload( + ExecutableStagePayload.newBuilder() + .setInput("input-pc") + .setEnvironment(environment) + .setComponents( + Components.newBuilder() + .putPcollections( + "input-pc", + PCollection.newBuilder() + .setWindowingStrategyId( + "windowing-strategy") + .setCoderId("coder-id") + .build()) + .putWindowingStrategies( + "windowing-strategy", + WindowingStrategy.newBuilder() + .setWindowCoderId("coder-id") + .build()) + .putCoders( + "coder-id", + Coder.newBuilder() + .setSpec( + FunctionSpec.newBuilder() + .setUrn( + ModelCoders + .INTERVAL_WINDOW_CODER_URN) + .build()) + .build()) + .build()) + .build()); + } +} diff --git a/flink-python/src/test/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManagerTest.java b/flink-python/src/test/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManagerTest.java index bcd9b63d2cc..68ee4ccadb0 100644 --- a/flink-python/src/test/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManagerTest.java +++ b/flink-python/src/test/java/org/apache/flink/python/env/process/ProcessPythonEnvironmentManagerTest.java @@ -53,6 +53,7 @@ import static org.apache.flink.python.env.process.ProcessPythonEnvironmentManage import static org.apache.flink.python.env.process.ProcessPythonEnvironmentManager.PYTHON_REQUIREMENTS_FILE; import static org.apache.flink.python.env.process.ProcessPythonEnvironmentManager.PYTHON_REQUIREMENTS_INSTALL_DIR; import static org.apache.flink.python.env.process.ProcessPythonEnvironmentManager.PYTHON_WORKING_DIR; +import static org.apache.flink.python.util.PythonEnvironmentManagerUtils.PYFLINK_UDF_RUNNER_DIR; import static org.assertj.core.api.Assertions.assertThat; /** Tests for the {@link ProcessPythonEnvironmentManager}. */ @@ -229,6 +230,44 @@ class ProcessPythonEnvironmentManagerTest { } } + @Test + void testProcessEnvironmentOnlyContainsManagedVariables() throws Exception { + Map<String, String> pythonFiles = new HashMap<>(); + pythonFiles.put(String.join(File.separator, tmpDir, "file1"), "test_file1.py"); + PythonDependencyInfo dependencyInfo = + new PythonDependencyInfo(pythonFiles, null, null, new HashMap<>(), "python"); + // what the TaskManager process would pass on + Map<String, String> systemEnv = new HashMap<>(); + systemEnv.put(PYFLINK_UDF_RUNNER_DIR, ""); + systemEnv.put("PYTHONPATH", "/inherited/pythonpath"); + systemEnv.put("INHERITED_SECRET", "must-not-be-passed-on"); + + try (ProcessPythonEnvironmentManager environmentManager = + new ProcessPythonEnvironmentManager( + dependencyInfo, new String[] {tmpDir}, systemEnv, new JobID())) { + environmentManager.open(); + String baseDir = environmentManager.getBaseDirectory(); + + ProcessPythonEnvironment environment = + (ProcessPythonEnvironment) environmentManager.createEnvironment(); + + // the worker inherits the process environment; only what the manager set is passed on + assertThat(environment.getEnv()) + .doesNotContainKeys("INHERITED_SECRET", PYFLINK_UDF_RUNNER_DIR) + .containsEntry( + "PYTHONPATH", + String.join( + File.pathSeparator, + String.join(File.separator, baseDir, PYTHON_FILES_DIR, "file1"), + "/inherited/pythonpath")) + .containsEntry("python", "python") + .containsEntry("BOOT_LOG_DIR", baseDir) + .containsEntry(PYFLINK_GATEWAY_DISABLED, "true"); + // the derivation itself still sees the whole environment + assertThat(environmentManager.getPythonEnv()).containsKey("INHERITED_SECRET"); + } + } + @Test void testRequirements() throws Exception { PythonDependencyInfo dependencyInfo =
