From d63eb5157e09ba5ca0001532f23a38c731c75479 Mon Sep 17 00:00:00 2001 From: Reuven Lax Date: Mon, 28 Sep 2026 11:24:58 -0700 Subject: [PATCH 1/2] Tune Dataflow ValidatesRunner worker options, class filtering, and category exclusions --- .../google-cloud-dataflow-java/build.gradle | 40 ++++++++++++++++++- .../beam/sdk/transforms/GroupByKeyTest.java | 2 - .../apache/beam/sdk/transforms/ParDoTest.java | 17 ++++++-- 3 files changed, 52 insertions(+), 7 deletions(-) diff --git a/runners/google-cloud-dataflow-java/build.gradle b/runners/google-cloud-dataflow-java/build.gradle index fbfa837d9fd6..5c2d0a8befca 100644 --- a/runners/google-cloud-dataflow-java/build.gradle +++ b/runners/google-cloud-dataflow-java/build.gradle @@ -174,6 +174,9 @@ def legacyPipelineOptions = [ "--region=${gcpRegion}", "--tempRoot=${dataflowValidatesTempRoot}", "--dataflowWorkerJar=${dataflowLegacyWorkerJar}", + "--numWorkers=1", + "--maxNumWorkers=1", + "--diskSizeGb=30", "--usePublicIps=false", "--experiments=enable_lineage" ] @@ -192,6 +195,9 @@ def runnerV2CommonPipelineOptions = [ "--tempRoot=${dataflowValidatesTempRoot}", "--experiments=use_unified_worker,use_runner_v2", "--firestoreDb=${firestoreDb}", + "--numWorkers=1", + "--maxNumWorkers=1", + "--diskSizeGb=30", "--usePublicIps=false", "--experiments=enable_lineage" ] @@ -228,6 +234,29 @@ def commonRunnerV2ExcludeCategories = [ 'org.apache.beam.sdk.testing.UsesBoundedTrieMetrics', // Dataflow QM as of now does not support returning back BoundedTrie in metric result. ] +def isValidatesRunnerTestClass = { FileTreeElement element -> + if (element.isDirectory()) { + return true + } + if (!element.name.endsWith('.class') || + element.name.startsWith('ValidateRunnerXlangTest')) { + return false + } + def content = new String(element.file.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) + if (content.contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;')) { + return true + } + if (content.contains('Lorg/junit/experimental/runners/Enclosed;')) { + def prefix = element.name.substring(0, element.name.length() - 6) + '$' + return element.file.parentFile.listFiles()?.any { File f -> + f.name.startsWith(prefix) && f.name.endsWith('.class') && + new String(f.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) + .contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;') + } ?: false + } + return false +} + def createLegacyWorkerValidatesRunnerTest = { Map args -> def name = args.name def pipelineOptions = args.pipelineOptions ?: legacyPipelineOptions @@ -245,6 +274,7 @@ def createLegacyWorkerValidatesRunnerTest = { Map args -> classpath = configurations.validatesRunner testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs) + files(project(project.path).sourceSets.test.output.classesDirs) + include isValidatesRunnerTestClass useJUnit { includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner' commonLegacyExcludeCategories.each { @@ -278,6 +308,7 @@ def createRunnerV2ValidatesRunnerTest = { Map args -> classpath = configurations.validatesRunner testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs) + files(project(project.path).sourceSets.test.output.classesDirs) + include isValidatesRunnerTestClass useJUnit { includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner' commonRunnerV2ExcludeCategories.each { @@ -466,8 +497,13 @@ task validatesRunner { 'org.apache.beam.sdk.transforms.ParDoLifecycleTest.testTeardownCalledAfterExceptionInStartBundle', 'org.apache.beam.sdk.transforms.ParDoLifecycleTest.testTeardownCalledAfterExceptionInStartBundleStateful', ], - // Batch legacy worker does not support bundle finalization. - excludedCategories: [ 'org.apache.beam.sdk.testing.UsesBundleFinalizer', ], + // Batch legacy worker does not support bundle finalization, triggered side inputs, or unbounded PCollections. + excludedCategories: [ + 'org.apache.beam.sdk.testing.UsesBundleFinalizer', + 'org.apache.beam.sdk.testing.UsesTriggeredSideInputs', + 'org.apache.beam.sdk.testing.UsesUnboundedPCollections', + 'org.apache.beam.sdk.testing.UsesUnboundedSplittableParDo', + ], )) } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java index 5464838ad4db..18541437d5fb 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java @@ -94,7 +94,6 @@ import org.junit.Rule; import org.junit.Test; import org.junit.experimental.categories.Category; -import org.junit.experimental.runners.Enclosed; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -104,7 +103,6 @@ "unchecked", "unused" }) -@RunWith(Enclosed.class) public class GroupByKeyTest implements Serializable { /** Shared test base class with setup/teardown helpers. */ public abstract static class SharedTestBase { diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java index 6beea338689b..7eb704b2bbfb 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java @@ -3356,7 +3356,8 @@ public void testOrderedListStateBounded() { ValidatesRunner.class, UsesStatefulParDo.class, UsesOrderedListState.class, - UsesOnWindowExpiration.class + UsesOnWindowExpiration.class, + UsesUnboundedPCollections.class }) public void testOrderedListStateUnbounded() { testOrderedListStateImpl(true); @@ -3420,7 +3421,12 @@ public void testOrderedListStateRangeFetchBounded() { } @Test - @Category({ValidatesRunner.class, UsesStatefulParDo.class, UsesOrderedListState.class}) + @Category({ + ValidatesRunner.class, + UsesStatefulParDo.class, + UsesOrderedListState.class, + UsesUnboundedPCollections.class + }) public void testOrderedListStateRangeFetchUnbounded() { testOrderedListStateRangeFetchImpl(true); } @@ -3493,7 +3499,12 @@ public void testOrderedListStateRangeDeleteBounded() { } @Test - @Category({ValidatesRunner.class, UsesStatefulParDo.class, UsesOrderedListState.class}) + @Category({ + ValidatesRunner.class, + UsesStatefulParDo.class, + UsesOrderedListState.class, + UsesUnboundedPCollections.class + }) public void testOrderedListStateRangeDeleteUnbounded() { testOrderedListStateRangeDeleteImpl(true); } From 3d7efef614e898198d7b47afad2a92dfd8c033c9 Mon Sep 17 00:00:00 2001 From: Reuven Lax Date: Mon, 28 Sep 2026 12:20:16 -0700 Subject: [PATCH 2/2] remove old code --- runners/google-cloud-dataflow-java/build.gradle | 15 ++------------- 1 file changed, 2 insertions(+), 13 deletions(-) diff --git a/runners/google-cloud-dataflow-java/build.gradle b/runners/google-cloud-dataflow-java/build.gradle index 5c2d0a8befca..d52d2f1bc7b3 100644 --- a/runners/google-cloud-dataflow-java/build.gradle +++ b/runners/google-cloud-dataflow-java/build.gradle @@ -242,19 +242,8 @@ def isValidatesRunnerTestClass = { FileTreeElement element -> element.name.startsWith('ValidateRunnerXlangTest')) { return false } - def content = new String(element.file.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) - if (content.contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;')) { - return true - } - if (content.contains('Lorg/junit/experimental/runners/Enclosed;')) { - def prefix = element.name.substring(0, element.name.length() - 6) + '$' - return element.file.parentFile.listFiles()?.any { File f -> - f.name.startsWith(prefix) && f.name.endsWith('.class') && - new String(f.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) - .contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;') - } ?: false - } - return false + return new String(element.file.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) + .contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;') } def createLegacyWorkerValidatesRunnerTest = { Map args ->