You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by sh...@apache.org on 2026/06/26 16:14:19 UTC
(beam) branch debug-flink-load-test created (now 3009747717c)
This is an automated email from the ASF dual-hosted git repository.
shunping pushed a change to branch debug-flink-load-test
in repository https://gitbox.apache.org/repos/asf/beam.git
at 3009747717c Attempt to fix.
This branch includes the following new commits:
new 3009747717c Attempt to fix.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 01/01: Attempt to fix.
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch debug-flink-load-test
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 3009747717c9d63fc9c0673b696dd746893841be
Author: Shunping Huang <sh...@google.com>
AuthorDate: Fri Jun 26 12:13:20 2026 -0400
Attempt to fix.
---
.test-infra/dataproc/init-actions/flink.sh | 12 ++++++--
.../flink/FlinkExecutionEnvironmentsTest.java | 3 ++
.../beam/runners/flink/FlinkSubmissionTest.java | 32 +++++++++++++---------
.../dataproc/dataproc_cluster_manager.py | 9 ++++--
4 files changed, 38 insertions(+), 18 deletions(-)
diff --git a/.test-infra/dataproc/init-actions/flink.sh b/.test-infra/dataproc/init-actions/flink.sh
index 7d409e6dc60..934355fa6e3 100644
--- a/.test-infra/dataproc/init-actions/flink.sh
+++ b/.test-infra/dataproc/init-actions/flink.sh
@@ -149,11 +149,16 @@ function configure_flink() {
# create working directory
mkdir -p "${FLINK_WORKING_DIR}"
+ local config_file="${FLINK_INSTALL_DIR}/conf/flink-conf.yaml"
+ if [[ -f "${FLINK_INSTALL_DIR}/conf/config.yaml" ]]; then
+ config_file="${FLINK_INSTALL_DIR}/conf/config.yaml"
+ fi
+
# Apply Flink settings by appending them to the default config.
- cat << EOF >> ${FLINK_INSTALL_DIR}/conf/flink-conf.yaml
+ cat << EOF >> "${config_file}"
# Settings applied by Cloud Dataproc initialization action
jobmanager.rpc.address: ${master_hostname}
-jobmanager.heap.mb: ${flink_jobmanager_memory}
+jobmanager.memory.process.size: "${flink_jobmanager_memory} mb"
taskmanager.memory.process.size: "${flink_taskmanager_memory} mb"
taskmanager.memory.jvm-metaspace.size: 512 mb
taskmanager.memory.task.off-heap.size: 256 mb
@@ -161,7 +166,8 @@ taskmanager.memory.managed.fraction: 0.5
taskmanager.numberOfTaskSlots: ${flink_taskmanager_slots}
parallelism.default: ${flink_parallelism}
fs.hdfs.hadoopconf: ${HADOOP_CONF_DIR}
-state.backend: filesystem
+state.backend.type: hashmap
+state.checkpoint-storage: filesystem
state.checkpoints.dir: ${checkpoints_dir}
EOF
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java
index 7262760a632..d31a856cb9f 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java
@@ -394,6 +394,9 @@ public class FlinkExecutionEnvironmentsTest {
InputStream inputStream = getClass().getResourceAsStream("/flink-conf.yaml");
File root = temporaryFolder.getRoot();
Files.copy(inputStream, new File(root, "flink-conf.yaml").toPath());
+ try (InputStream inputStream2 = getClass().getResourceAsStream("/flink-conf.yaml")) {
+ Files.copy(inputStream2, new File(root, "config.yaml").toPath());
+ }
return root.getAbsolutePath();
}
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java
index 8e4c3255fac..6678ef56062 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java
@@ -19,7 +19,6 @@ package org.apache.beam.runners.flink;
import java.io.File;
import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.security.Permission;
@@ -211,6 +210,8 @@ public class FlinkSubmissionTest {
flinkCluster.getRestPort());
Files.write(file.toPath(), config.getBytes(StandardCharsets.UTF_8));
+ File configYaml = new File(file.getParent(), "config.yaml");
+ Files.write(configYaml.toPath(), config.getBytes(StandardCharsets.UTF_8));
// Create a new environment with the location of the Flink config for CliFrontend
ImmutableMap<String, String> newEnv =
@@ -232,19 +233,24 @@ public class FlinkSubmissionTest {
* be set using the {@code ConfigConstants.ENV_FLINK_CONF_DIR} environment variable.
*/
private static void modifyEnv(Map<String, String> env) throws Exception {
- Class processEnv = Class.forName("java.lang.ProcessEnvironment");
- Field envField = processEnv.getDeclaredField("theUnmodifiableEnvironment");
+ Class<?> processEnvClass = Class.forName("java.lang.ProcessEnvironment");
+ Field theEnvironmentField = processEnvClass.getDeclaredField("theEnvironment");
+ theEnvironmentField.setAccessible(true);
+ Map<String, String> envMap = (Map<String, String>) theEnvironmentField.get(null);
+ envMap.clear();
+ envMap.putAll(env);
- Field modifiersField = Field.class.getDeclaredField("modifiers");
- modifiersField.setAccessible(true);
- modifiersField.setInt(envField, envField.getModifiers() & ~Modifier.FINAL);
-
- envField.setAccessible(true);
- envField.set(null, env);
- envField.setAccessible(false);
-
- modifiersField.setInt(envField, envField.getModifiers() & Modifier.FINAL);
- modifiersField.setAccessible(false);
+ try {
+ Field theCaseInsensitiveEnvironmentField =
+ processEnvClass.getDeclaredField("theCaseInsensitiveEnvironment");
+ theCaseInsensitiveEnvironmentField.setAccessible(true);
+ Map<String, String> cienvMap =
+ (Map<String, String>) theCaseInsensitiveEnvironmentField.get(null);
+ cienvMap.clear();
+ cienvMap.putAll(env);
+ } catch (NoSuchFieldException e) {
+ // Ignore if not on Windows
+ }
}
/** Prevents the CliFrontend from calling System.exit. */
diff --git a/sdks/python/apache_beam/runners/interactive/dataproc/dataproc_cluster_manager.py b/sdks/python/apache_beam/runners/interactive/dataproc/dataproc_cluster_manager.py
index 817d4f6b583..cb77da55509 100644
--- a/sdks/python/apache_beam/runners/interactive/dataproc/dataproc_cluster_manager.py
+++ b/sdks/python/apache_beam/runners/interactive/dataproc/dataproc_cluster_manager.py
@@ -59,14 +59,19 @@ sudo -u yarn gcloud auth configure-docker --quiet
readonly FLINK_INSTALL_DIR='/usr/lib/flink'
readonly MASTER_HOSTNAME="$(/usr/share/google/get_metadata_value attributes/dataproc-master)"
-cat <<EOF >>${FLINK_INSTALL_DIR}/conf/flink-conf.yaml
+config_file="${FLINK_INSTALL_DIR}/conf/flink-conf.yaml"
+if [[ -f "${FLINK_INSTALL_DIR}/conf/config.yaml" ]]; then
+ config_file="${FLINK_INSTALL_DIR}/conf/config.yaml"
+fi
+
+cat <<EOF >>"${config_file}"
taskmanager.memory.network.fraction: 0.2
taskmanager.memory.network.min: 64mb
taskmanager.memory.network.max: 1gb
EOF
sed -i \
"s/^taskmanager.network.numberOfBuffers: 2048/taskmanager.network.numberOfBuffers: 8192/" \
- ${FLINK_INSTALL_DIR}/conf/flink-conf.yaml
+ "${config_file}"
if [[ "${HOSTNAME}" == "${MASTER_HOSTNAME}" ]]; then
. /usr/bin/flink-yarn-daemon