From cc207fc8f5041128951a79bdc54d2095fc98c757 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Sat, 24 Oct 2026 17:32:49 +0000 Subject: [PATCH 1/2] [#36841] Spark 4 streaming: ValidatesRunner suite for the Structured Streaming runner --- ..._Spark4_StructuredStreaming_Streaming.json | 3 + .github/workflows/README.md | 1 + ...r_Spark4_StructuredStreaming_Streaming.yml | 89 +++++++++++++++++++ runners/spark/4/build.gradle | 76 ++++++++++++++++ runners/spark/spark_runner.gradle | 31 ++++--- .../translation/SparkTransformOverrides.java | 5 ++ 6 files changed, 192 insertions(+), 13 deletions(-) create mode 100644 .github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json create mode 100644 .github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json new file mode 100644 index 000000000000..c4edaa85a89d --- /dev/null +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json @@ -0,0 +1,3 @@ +{ + "comment": "Modify this file in a trivial way to cause this test suite to run" +} diff --git a/.github/workflows/README.md b/.github/workflows/README.md index 682b51699c09..7026f2d6604b 100644 --- a/.github/workflows/README.md +++ b/.github/workflows/README.md @@ -356,6 +356,7 @@ PostCommit Jobs run in a schedule against master branch and generally do not get | [ PostCommit Java ValidatesRunner Flink ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Flink.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner Spark ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Spark.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner SparkStructuredStreaming ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml?query=event%3Aschedule) | +| [ PostCommit Java ValidatesRunner Spark4 StructuredStreaming Streaming ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner Spark4 ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Spark4.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner ULR ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_ULR.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml?query=event%3Aschedule) | | [ PostCommit Java ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java.yml) | N/A |`beam_PostCommit_Java.json`| [![.github/workflows/beam_PostCommit_Java.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java.yml?query=event%3Aschedule) | diff --git a/.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml b/.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml new file mode 100644 index 000000000000..6f5e55020682 --- /dev/null +++ b/.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml @@ -0,0 +1,89 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +name: PostCommit Java ValidatesRunner Spark4 StructuredStreaming Streaming + +on: + schedule: + - cron: '45 4/6 * * *' + pull_request: + paths: ['release/trigger_all_tests.json', '.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json'] + workflow_dispatch: + +# Setting explicit permissions for the action +permissions: + actions: write + pull-requests: write + checks: write + contents: read + deployments: read + id-token: none + issues: write + discussions: read + packages: read + pages: read + repository-projects: read + security-events: read + statuses: read + +# This allows a subsequently queued workflow run to interrupt previous runs +concurrency: + group: '${{ github.workflow }} @ ${{ github.event.pull_request.number || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.sender.login }}' + cancel-in-progress: true + +env: + DEVELOCITY_ACCESS_KEY: ${{ secrets.BEAM_DEVELOCITY_ACCESS_KEY }} + +jobs: + beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming: + name: ${{ matrix.job_name }} + runs-on: [self-hosted, ubuntu-24.04, main] + timeout-minutes: 120 + strategy: + matrix: + job_name: [beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming] + if: | + github.event_name == 'workflow_dispatch' || + github.event_name == 'pull_request' || + (github.event_name == 'schedule' && github.repository == 'apache/beam') + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + - name: Setup repository + uses: ./.github/actions/setup-action + - name: Setup environment + uses: ./.github/actions/setup-environment-action + with: + java-version: '17' + - name: run validatesStructuredStreamingRunnerStreaming script + uses: ./.github/actions/gradle-command-self-hosted-action + with: + gradle-command: :runners:spark:4:validatesStructuredStreamingRunnerStreaming + - name: Archive JUnit Test Results + uses: actions/upload-artifact@v7 + if: ${{ !success() }} + with: + name: JUnit Test Results + path: "**/build/reports/tests/" + - name: Publish JUnit Test Results + uses: EnricoMi/publish-unit-test-result-action@v2 + if: always() + with: + commit: '${{ env.GITHUB_SHA }}' + comment_mode: 'off' + check_run: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }} + files: '**/build/test-results/**/*.xml' + large_files: true 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..513c442f6123 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 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), From 65c3494dd7c0002c61d173979810b6ee2b1d41c6 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Tue, 3 Nov 2026 21:28:12 +0000 Subject: [PATCH 2/2] Run the streaming suite from the Spark 4 validatesRunner task instead of a separate workflow --- ...ostCommit_Java_ValidatesRunner_Spark4.json | 2 +- ..._Spark4_StructuredStreaming_Streaming.json | 3 - .github/workflows/README.md | 1 - ...r_Spark4_StructuredStreaming_Streaming.yml | 89 ------------------- runners/spark/spark_runner.gradle | 3 + 5 files changed, 4 insertions(+), 94 deletions(-) delete mode 100644 .github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json delete mode 100644 .github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml 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/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json deleted file mode 100644 index c4edaa85a89d..000000000000 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json +++ /dev/null @@ -1,3 +0,0 @@ -{ - "comment": "Modify this file in a trivial way to cause this test suite to run" -} diff --git a/.github/workflows/README.md b/.github/workflows/README.md index 7026f2d6604b..682b51699c09 100644 --- a/.github/workflows/README.md +++ b/.github/workflows/README.md @@ -356,7 +356,6 @@ PostCommit Jobs run in a schedule against master branch and generally do not get | [ PostCommit Java ValidatesRunner Flink ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Flink.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner Spark ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Spark.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner SparkStructuredStreaming ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml?query=event%3Aschedule) | -| [ PostCommit Java ValidatesRunner Spark4 StructuredStreaming Streaming ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner Spark4 ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_Spark4.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml?query=event%3Aschedule) | | [ PostCommit Java ValidatesRunner ULR ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml) | N/A |`beam_PostCommit_Java_ValidatesRunner_ULR.json`| [![.github/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_ULR.yml?query=event%3Aschedule) | | [ PostCommit Java ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java.yml) | N/A |`beam_PostCommit_Java.json`| [![.github/workflows/beam_PostCommit_Java.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java.yml?query=event%3Aschedule) | diff --git a/.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml b/.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml deleted file mode 100644 index 6f5e55020682..000000000000 --- a/.github/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.yml +++ /dev/null @@ -1,89 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -name: PostCommit Java ValidatesRunner Spark4 StructuredStreaming Streaming - -on: - schedule: - - cron: '45 4/6 * * *' - pull_request: - paths: ['release/trigger_all_tests.json', '.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming.json'] - workflow_dispatch: - -# Setting explicit permissions for the action -permissions: - actions: write - pull-requests: write - checks: write - contents: read - deployments: read - id-token: none - issues: write - discussions: read - packages: read - pages: read - repository-projects: read - security-events: read - statuses: read - -# This allows a subsequently queued workflow run to interrupt previous runs -concurrency: - group: '${{ github.workflow }} @ ${{ github.event.pull_request.number || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.sender.login }}' - cancel-in-progress: true - -env: - DEVELOCITY_ACCESS_KEY: ${{ secrets.BEAM_DEVELOCITY_ACCESS_KEY }} - -jobs: - beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming: - name: ${{ matrix.job_name }} - runs-on: [self-hosted, ubuntu-24.04, main] - timeout-minutes: 120 - strategy: - matrix: - job_name: [beam_PostCommit_Java_ValidatesRunner_Spark4_StructuredStreaming_Streaming] - if: | - github.event_name == 'workflow_dispatch' || - github.event_name == 'pull_request' || - (github.event_name == 'schedule' && github.repository == 'apache/beam') - steps: - - uses: actions/checkout@v7 - with: - persist-credentials: false - - name: Setup repository - uses: ./.github/actions/setup-action - - name: Setup environment - uses: ./.github/actions/setup-environment-action - with: - java-version: '17' - - name: run validatesStructuredStreamingRunnerStreaming script - uses: ./.github/actions/gradle-command-self-hosted-action - with: - gradle-command: :runners:spark:4:validatesStructuredStreamingRunnerStreaming - - name: Archive JUnit Test Results - uses: actions/upload-artifact@v7 - if: ${{ !success() }} - with: - name: JUnit Test Results - path: "**/build/reports/tests/" - - name: Publish JUnit Test Results - uses: EnricoMi/publish-unit-test-result-action@v2 - if: always() - with: - commit: '${{ env.GITHUB_SHA }}' - comment_mode: 'off' - check_run: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }} - files: '**/build/test-results/**/*.xml' - large_files: true diff --git a/runners/spark/spark_runner.gradle b/runners/spark/spark_runner.gradle index 513c442f6123..0569c09d1228 100644 --- a/runners/spark/spark_runner.gradle +++ b/runners/spark/spark_runner.gradle @@ -565,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