From 3c86ba818d88e4e139f2ce1dd825276fd5634a85 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Tue, 6 Oct 2026 09:39:43 +0000 Subject: [PATCH] [Spark] Run ValidatesRunner tests serially, runner metrics are JVM wide Fixes #40427. --- .../beam_PostCommit_Java_PVR_Spark3_Streaming.json | 2 +- .../trigger_files/beam_PostCommit_Java_PVR_Spark4_Batch.json | 3 ++- .../beam_PostCommit_Java_PVR_Spark4_Streaming.json | 3 ++- .../beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json | 2 +- .../trigger_files/beam_PostCommit_Java_PVR_Spark_Batch.json | 2 +- .../beam_PostCommit_Java_ValidatesRunner_Spark.json | 2 +- .../beam_PostCommit_Java_ValidatesRunner_Spark4.json | 2 +- ...Commit_Java_ValidatesRunner_SparkStructuredStreaming.json | 3 ++- runners/spark/job-server/spark_job_server.gradle | 5 +++++ runners/spark/spark_runner.gradle | 5 +++++ 10 files changed, 21 insertions(+), 8 deletions(-) diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json index cc664e5ed96f..219fe098b669 100644 --- a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json +++ b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 9, + "modification": 10, "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" } diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Batch.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Batch.json index 04197c72efed..441fc7a0f755 100644 --- a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Batch.json +++ b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Batch.json @@ -1,4 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" + "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner", + "https://github.com/apache/beam/issues/40427": "Spark ValidatesRunner tests run serially" } diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Streaming.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Streaming.json index 04197c72efed..441fc7a0f755 100644 --- a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Streaming.json +++ b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_Streaming.json @@ -1,4 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" + "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner", + "https://github.com/apache/beam/issues/40427": "Spark ValidatesRunner tests run serially" } diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json index 5abe02fc09c7..3a009261f4f9 100644 --- a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json +++ b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 1 + "modification": 2 } diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark_Batch.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark_Batch.json index 975b99265d4e..c1bb7d6c3798 100644 --- a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark_Batch.json +++ b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark_Batch.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 7, + "modification": 8, "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" } diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark.json index 62ae7886c573..2160d3c68005 100644 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark.json +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 4 + "modification": 5 } diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json index b26833333238..c537844dc84a 100644 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 2 + "modification": 3 } diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json index 373c31ff2341..ea9b041f3fc0 100644 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json @@ -9,5 +9,6 @@ "https://github.com/apache/beam/pull/34155": "noting that PR #34155 should run this test", "https://github.com/apache/beam/pull/35159": "moving WindowedValue and making an interface", "https://github.com/apache/beam/pull/39793": "noting that PR #39793 should run this test", - "https://github.com/apache/beam/pull/40103": "noting that PR #40103 should run this test" + "https://github.com/apache/beam/pull/40103": "noting that PR #40103 should run this test", + "https://github.com/apache/beam/issues/40427": "Spark ValidatesRunner tests run serially" } diff --git a/runners/spark/job-server/spark_job_server.gradle b/runners/spark/job-server/spark_job_server.gradle index 637ac98375da..7e5e65aa4c40 100644 --- a/runners/spark/job-server/spark_job_server.gradle +++ b/runners/spark/job-server/spark_job_server.gradle @@ -94,6 +94,11 @@ def sickbayTests = [ 'org.apache.beam.sdk.transforms.ReshuffleTest.testReshufflePreservesMetadata', ] +// One pipeline per JVM at a time, Spark runner metrics and the SparkContext are JVM wide. +tasks.withType(Test).configureEach { + systemProperty "beam.validatesRunner.parallelThreads", "1" +} + def portableValidatesRunnerTask(String name, boolean streaming, boolean docker, ArrayList sickbayTests, boolean structuredStreaming = false) { def pipelineOptions = [] if (structuredStreaming) { diff --git a/runners/spark/spark_runner.gradle b/runners/spark/spark_runner.gradle index 2a161db0a820..c40eb788b61c 100644 --- a/runners/spark/spark_runner.gradle +++ b/runners/spark/spark_runner.gradle @@ -218,6 +218,11 @@ test { if(project.hasProperty("rerun-tests")) { outputs.upToDateWhen {false} } } +// One pipeline per JVM at a time, Spark runner metrics and the SparkContext are JVM wide. +tasks.withType(Test).configureEach { + systemProperty "beam.validatesRunner.parallelThreads", "1" +} + class SparkComponents { List components }