diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json index 455144f02a35..d6a91b7e2e86 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": 6 + "modification": 7 } diff --git a/runners/spark/4/build.gradle b/runners/spark/4/build.gradle index ae8750ec5b11..7ae41e473e24 100644 --- a/runners/spark/4/build.gradle +++ b/runners/spark/4/build.gradle @@ -51,6 +51,82 @@ tasks.validatesStructuredStreamingRunnerBatch { } } +tasks.register("validatesStructuredStreamingRunnerStreaming", Test) { + applyStructuredStreamingValidatesRunnerSetup(it, ["--streaming": "true", "--streamingStopAfterIdleBatches": "5"], [ + // Side inputs + 'org.apache.beam.sdk.testing.UsesSideInputs', + // Unbounded + 'org.apache.beam.sdk.testing.UsesTestStreamWithProcessingTime', + // State and Timers + 'org.apache.beam.sdk.testing.UsesProcessingTimeTimers', + 'org.apache.beam.sdk.testing.UsesLoopingTimer', + 'org.apache.beam.sdk.testing.UsesMultimapState', + // Windowing + 'org.apache.beam.sdk.testing.UsesCustomWindowMerging', + // Ordering + 'org.apache.beam.sdk.testing.UsesRequiresTimeSortedInput', + // SDF + 'org.apache.beam.sdk.testing.UsesBoundedSplittableParDo', + // Exceeds Java heap space + 'org.apache.beam.sdk.testing.LargeKeys$Above100MB', + ]) + + filter { + // TODO(https://github.com/apache/beam/issues/36841): merging windows + excludeTestsMatching 'org.apache.beam.sdk.transforms.CombineTest$WindowingTests.testSessionsCombine' + excludeTestsMatching 'org.apache.beam.sdk.transforms.GroupByKeyTest$WindowTests.testGroupByKeyMergingWindows' + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterSessionsAndGroupByKey' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ReshuffleTest.testReshuffleAfterSessionsAndGroupByKey' + excludeTestsMatching 'org.apache.beam.sdk.transforms.windowing.WindowingTest.testMergingWindowing' + + // TODO(https://github.com/apache/beam/issues/36841): side inputs, PAssert.thatSingleton and Combine.globally with defaults + excludeTestsMatching 'org.apache.beam.sdk.io.CountingSourceTest.testBoundedSourceSplits' + excludeTestsMatching 'org.apache.beam.sdk.io.CountingSourceTest.testUnboundedSourceSplits' + excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testIsEqualTo' + excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testNotEqualTo' + excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testWindowedIsEqualTo' + excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFalse' + excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFalseDefaultReasonString' + excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFailsForEmptyPCollection' + excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFailsForNonSingletonPCollection' + + // TODO(https://github.com/apache/beam/issues/40949): ParDo with additional outputs + excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAllAttemptedMetrics' + excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedBoundedTrieMetrics' + excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedCounterMetrics' + excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedDistributionMetrics' + excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedGaugeMetrics' + excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedStringSetMetrics' + excludeTestsMatching 'org.apache.beam.sdk.transforms.CombineTest$BasicTests.testHotKeyCombining' + excludeTestsMatching 'org.apache.beam.sdk.transforms.CombineTest$BasicTests.testHotKeyCombiningWithAccumulationMode' + excludeTestsMatching 'org.apache.beam.sdk.transforms.FlattenTest.testFlattenMultiplePCollectionsHavingMultipleConsumers' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoSchemaTest.testReadAndWriteMultiOutput' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoEmptyWithTaggedOutput' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoWithEmptyTaggedOutput' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoWithOnlyTaggedOutput' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoWithTaggedOutput' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$StateTests.testValueStateTaggedOutput' + excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$TimestampTests.testParDoTaggedOutputWithTimestamp' + excludeTestsMatching 'org.apache.beam.sdk.values.PCollectionTupleTest.testComposePCollectionTuple' + + // TODO(https://github.com/apache/beam/issues/40950): Redistribute expands to GroupByKey with ReshuffleTrigger + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testJustRedistribute' + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterFixedWindows' + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterFixedWindowsAndGroupByKey' + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterSlidingWindows' + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterSlidingWindowsAndGroupByKey' + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributePreservesMetadata' + excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributePreservesTimestamps' + + // TODO(https://github.com/apache/beam/issues/36841): PeriodicImpulse uses a splittable DoFn + excludeTestsMatching 'org.apache.beam.sdk.transforms.PerKeyOrderingTest.testMultipleStatefulOrderingWithShuffle' + excludeTestsMatching 'org.apache.beam.sdk.transforms.PerKeyOrderingTest.testMultipleStatefulOrderingWithoutShuffle' + + // TODO(https://github.com/apache/beam/issues/32021) + excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testBoundedSourceMetricsInSplit' + } +} + tasks.validatesRunner.dependsOn(validatesStructuredStreamingRunnerBatch) // Exclude legacy DStream-based streaming tests (under runners/spark/translation/streaming) diff --git a/runners/spark/spark_runner.gradle b/runners/spark/spark_runner.gradle index c40eb788b61c..0569c09d1228 100644 --- a/runners/spark/spark_runner.gradle +++ b/runners/spark/spark_runner.gradle @@ -486,33 +486,32 @@ def validatesRunnerStreaming = tasks.register("validatesRunnerStreaming", Test) } } -tasks.register("validatesStructuredStreamingRunnerBatch", Test) { - group = "Verification" +def applyStructuredStreamingValidatesRunnerSetup = { Test it, Map pipelineOptions, List extraExcludedCategories -> + it.group = "Verification" // Disable gradle cache - outputs.upToDateWhen { false } - systemProperties sparkTestProperties(["--runner":"SparkStructuredStreamingRunner", "--testMode":"true"]) + it.outputs.upToDateWhen { false } + it.systemProperties sparkTestProperties(["--runner":"SparkStructuredStreamingRunner", "--testMode":"true"] + pipelineOptions) // Register various other classes used in tests - systemProperty 'spark.kryo.classesToRegister', + it.systemProperty 'spark.kryo.classesToRegister', 'org.apache.beam.sdk.transforms.MapViewTest$NonDeterministicStringCoder,' + 'org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.RegularImmutableList' - jvmArgs += sparkTestJvmArgs() - jvmArgs '-Xmx7g' // Increase memory heap in order to avoid OOM errors + it.jvmArgs += sparkTestJvmArgs() + it.jvmArgs '-Xmx7g' // Increase memory heap in order to avoid OOM errors - classpath = configurations.validatesRunner - testClassesDirs = files( + it.classpath = project.configurations.validatesRunner + it.testClassesDirs = project.files( project(":sdks:java:core").sourceSets.test.output.classesDirs, project(":runners:core-java").sourceSets.test.output.classesDirs, ) - testClassesDirs += files(project.sourceSets.test.output.classesDirs) + it.testClassesDirs += project.files(project.sourceSets.test.output.classesDirs) - maxParallelForks 4 + it.maxParallelForks 4 - useJUnit { + it.useJUnit { includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner' // Should be run only in a properly configured SDK harness environment excludeCategories 'org.apache.beam.sdk.testing.UsesSdkHarnessEnvironment' // Unbounded - excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections' excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream' // State and Timers excludeCategories 'org.apache.beam.sdk.testing.UsesOnWindowExpiration' @@ -529,7 +528,13 @@ tasks.register("validatesStructuredStreamingRunnerBatch", Test) { excludeCategories 'org.apache.beam.sdk.testing.UsesExternalService' excludeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer' excludeCategories 'org.apache.beam.sdk.testing.UsesTriggeredSideInputs' + extraExcludedCategories.each { excludeCategories it } } +} +ext.applyStructuredStreamingValidatesRunnerSetup = applyStructuredStreamingValidatesRunnerSetup + +tasks.register("validatesStructuredStreamingRunnerBatch", Test) { + applyStructuredStreamingValidatesRunnerSetup(it, [:], ['org.apache.beam.sdk.testing.UsesUnboundedPCollections']) filter { // These build on PeriodicImpulse, so the pipeline is unbounded and rejected by this batch only // runner, but they are not categorized as UsesUnboundedPCollections. Excluded by name rather @@ -560,6 +565,9 @@ tasks.register("validatesRunner") { dependsOn validatesRunnerBatch dependsOn validatesRunnerBatchWithBoundedSDFExperiment dependsOn validatesRunnerStreaming + if (spark_major == '4') { + dependsOn "validatesStructuredStreamingRunnerStreaming" + } // It should be uncommented once all "validatesStructuredStreamingRunnerBatch" tests will pass. // Otherwise, it breaks Spark runner ValidatesRunner tests. //dependsOn validatesStructuredStreamingRunnerBatch diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/SparkTransformOverrides.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/SparkTransformOverrides.java index c53b89d4030a..63746cbe8165 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/SparkTransformOverrides.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/SparkTransformOverrides.java @@ -32,6 +32,7 @@ import org.apache.beam.sdk.transforms.Impulse; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.util.construction.EmptyFlattenAsCreateFactory; import org.apache.beam.sdk.util.construction.PTransformMatchers; import org.apache.beam.sdk.util.construction.PTransformTranslation; import org.apache.beam.sdk.util.construction.ReplacementOutputs; @@ -70,7 +71,11 @@ public static List getDefaultOverrides(boolean streaming) { PTransformMatchers.urnEqualTo(PTransformTranslation.SPLITTABLE_PROCESS_KEYED_URN), new SplittableParDoNaiveBounded.OverrideFactory())); } else { + // Before the Impulse override, the empty Flatten replacement expands to an Impulse. builder + .add( + PTransformOverride.of( + PTransformMatchers.emptyFlatten(), EmptyFlattenAsCreateFactory.instance())) .add( PTransformOverride.of( PTransformMatchers.classEqualTo(SplittableParDo.PrimitiveBoundedRead.class),