This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new cf7fbd3317d [FLINK-39572][python] Do not log the Beam environment
payload (#29225)
cf7fbd3317d is described below
commit cf7fbd3317dd9524096bf7d97580460935dc2501
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.
Generated-by: Claude Code (Claude Fable 5.1)
---
.../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 2e73dda2419..794ead6d849 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
@@ -227,7 +227,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);
}
@@ -698,7 +698,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?
}
@@ -712,7 +712,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 1652e51b4de..921ba3bfc39 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..105ca0272ec
--- /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.fnexecution.environment.EnvironmentFactory;
+import org.apache.beam.runners.fnexecution.environment.RemoteEnvironment;
+import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.apache.beam.sdk.util.construction.ModelCoders;
+import org.apache.beam.sdk.util.construction.graph.ExecutableStage;
+import org.apache.beam.vendor.grpc.v1p69p0.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 =
