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 <[email protected]> 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
