[runners-spark] Prep shared base for Spark 4 - #38324
Merged
Merged
Conversation
…, removed in Spark 4
- runners/spark/src/.../io/{SourceRDD,SparkUnboundedSource}.java:
scala.collection.JavaConversions -> JavaConverters (5 call-sites).
- runners/spark/src/.../stateful/SparkGroupAlsoByWindowViaWindowSet.java:
same JavaConversions -> JavaConverters migration (3 call-sites).
- runners/spark/src/.../translation/batch/DoFnRunnerFactory.java:
import scala.Serializable -> import java.io.Serializable (Scala 2.13
deprecated scala.Serializable; both are marker interfaces).
- runners/spark/src/.../translation/{SparkStreamingPortablePipelineTranslator,
streaming/StreamingTransformTranslator}.java:
StreamingContext.union(asScalaBuffer(dStreams)) ->
StreamingContext.union(asScalaBuffer(dStreams).toList()), since the
mutable-buffer overload of union was deprecated in Spark 3 and removed
in Spark 4 in favor of the immutable.Seq overload.
- runners/spark/src/.../translation/streaming/ParDoStateUpdateFn.java:
org.sparkproject.guava.collect.Iterators.emptyIterator() ->
java.util.Collections.emptyIterator() (Spark 4 removed the legacy
shaded-Guava Iterators).
- runners/spark/src/.../structuredstreaming/SparkStructuredStreamingPipelineResult.java:
static import of org.sparkproject.guava.base.Objects.firstNonNull ->
vendor.guava MoreObjects.firstNonNull (Spark 4 dropped the legacy
shaded Guava Objects class).
Behavior-identical on Spark 3.5 (every replacement is a Scala/Guava API
that exists with identical semantics on both Spark versions). First
split-out commit from apache#38255 per @Abacn's review guidance.
Also adds a CHANGES.md entry under "New Features / Improvements".
Three additive entries in buildSrc BeamModulePlugin.groovy: - def spark4_version = "4.0.2" alongside spark2_version and spark3_version. - project.ext.spark4_version export so per-project gradle scripts can reference it. - jackson_module_scala_2_13 library entry alongside the existing _2.11 and _2.12 entries. The first two are inert until the Spark 4 module landing PR (apache#38255) adds runners/spark/4/build.gradle that consumes them; the jackson_module_scala_2_13 library entry is independently useful for any future Scala 2.13 module. Second split-out commit from apache#38255 per @Abacn's review guidance.
…avaVersion to Spark 4 builds
runners/spark/spark_runner.gradle:
- Pass requireJavaVersion: (spark_version.startsWith("4")
? JavaVersion.VERSION_17 : null) into applyJavaNature so future
Spark 4 builds enforce Java 17. No-op for Spark 3.5 (returns null).
- Introduce isSparkAtLeast(minVersion) closure that compares
numerically, e.g. so "3.10.0" sorts after "3.5.0" instead of before
it lexicographically.
- Replace all 5 call-sites of `if ("$spark_version" >= "3.5.0")` with
`if (isSparkAtLeast("3.5.0"))`. Pure refactor, identical behaviour
on the currently-supported Spark 3.x range.
runners/spark/job-server/spark_job_server.gradle:
- Same requireJavaVersion: arg routing for the job-server, gated on
the parent spark_version property.
requireJavaVersion itself is pre-existing in BeamModulePlugin.groovy
(JavaNatureConfiguration parameter); this commit only adds invocation.
Third split-out commit from apache#38255 per @Abacn's review guidance.
Contributor
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request prepares the shared Spark runner base for upcoming Spark 4 compatibility. It focuses on refactoring code to be cross-compatible between Spark 3 and Spark 4, improving build infrastructure for version management, and ensuring correct Java version enforcement for future builds. These changes are behavior-identical on Spark 3.5 and serve as a foundational step for further Spark 4 integration. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
tkaymak
changed the title
tkaymak
changed the title
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request prepares the Spark runner for Spark 4 compatibility by migrating Scala collection conversions from JavaConversions to JavaConverters, replacing shaded Spark-Guava calls with Beam-vendor Guava or standard Java/Scala alternatives, and introducing a numeric isSparkAtLeast Gradle helper. It also enforces Java 17 for Spark 4 builds. The review feedback suggests improving the robustness of Gradle property lookups to avoid potential null pointer exceptions and ensuring consistent, safe version parsing within the new Gradle helper.
- spark_runner.gradle: move isSparkAtLeast closure above applyJavaNature
and use isSparkAtLeast("4.0.0") for the requireJavaVersion gate
instead of spark_version.startsWith("4"), for consistency and
robustness against future major version strings (e.g. "4.0.0-preview").
- spark_runner.gradle: parse minVersion the same way as spark_version
inside isSparkAtLeast — tokenize('.-').findAll { it.isInteger() } —
so a callsite using a suffixed version string can't trigger
NumberFormatException.
- spark_job_server.gradle: replace project.parent.findProperty (NPE if
the project has no parent) with project.findProperty, which already
walks up the project hierarchy. Renamed the local var to sparkVersion
to match the inheritance semantics.
Abacn
reviewed
Apr 29, 2026
Contributor
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
Contributor
|
Assigning reviewers: R: @Abacn for label build. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Abacn
approved these changes
Apr 30, 2026
tkaymak
added a commit
to tkaymak/beam
that referenced
this pull request
Apr 30, 2026
This was referenced Apr 30, 2026
tkaymak
added a commit
to tkaymak/beam
that referenced
this pull request
May 7, 2026
- SparkRunnerKryoRegistrator: throw IllegalStateException instead of
LOG.warn when neither ArraySeq$ofRef (Scala 2.13) nor WrappedArray$ofRef
(Scala 2.12) is on the classpath, so the missing class isn't silently
ignored. Drops the now-unused Logger field and slf4j imports.
- spark_runner.gradle: declare org.apache.spark:spark-connect-shims_2.13
as a provided dep gated on isSparkAtLeast("4.0.0"). Spark 4 splits the
Connect shim classes out of spark-sql; with enableStrictDependencies
this surfaced as analyzeClassesDependencies usedUndeclaredArtifacts.
The artifact does not exist for Spark 3, so the gate prevents Spark 3
resolution failures.
- runners/spark/4/build.gradle: drop the empty sparkVersions test
scaffolding (no additional Spark 4.x patch versions to test against
yet) and delete the now-unused
.github/workflows/beam_PreCommit_Java_Spark4_Versions.yml workflow
+ its README.md row.
- EncoderFactory (shared base): revert the line 94 switch to
STATIC_INVOKE_CONSTRUCTOR.getParameterCount(), keeping Spark 3 behavior
byte-for-byte unchanged. Spark 4's complete EncoderFactory override
under runners/spark/4/src/.../EncoderFactory.java is unaffected.
- CHANGES.md: drop the Highlights line for Spark 4. Will re-add when
ValidatesRunner tests are set up and confirmed working, matching the
Phase 1 apache#38324 pattern.
- runners/spark/4/job-server/container: delete the entire module
(build.gradle + Dockerfile) and remove its include() from
settings.gradle.kts. Per @Abacn's offer to defer the container module
to portable runner support later. The fat-jar :runners:spark:4:job-server
module is kept.
Abacn
pushed a commit
that referenced
this pull request
May 11, 2026
* build: add Spark 4.0.2 version property and Scala 2.13 support
Add spark4_version (4.0.2) to BeamModulePlugin alongside the existing
spark3_version. Update spark_runner.gradle to conditionally select the
correct Scala library (2.13 vs 2.12), Jackson module, Kafka test
dependency, and require Java 17 when building against Spark 4.
Register the new :runners:spark:4 module in settings.gradle.kts.
These changes are purely additive — all conditionals gate on
spark_version.startsWith("4") or spark_scala_version == '2.13', leaving
the Spark 3 build path untouched.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* refactor: make shared Spark source compatible with Scala 2.12 and 2.13
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* build: add runners/spark/4/ build configuration
Add the Gradle build file for the Spark 4 structured streaming runner.
The module mirrors runners/spark/3/ — it inherits the shared RDD-base
source from runners/spark/src/ via copySourceBase and adds its own
Structured Streaming implementation in src/main/java.
Key differences from the Spark 3 build:
- Uses spark4_version (4.0.2) with Scala 2.13.
- Excludes DStream-based streaming tests (Spark 4 supports only
structured streaming batch).
- Unconditionally adds --add-opens JVM flags required by Kryo on
Java 17 (Spark 4's minimum).
- Binds Spark driver to 127.0.0.1 for macOS compatibility.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* feat: add Spark 4 structured streaming runner source
Add the Spark 4 structured streaming runner implementation and tests.
Most files are adapted from the Spark 3 structured streaming runner
with targeted changes for Spark 4 / Scala 2.13 API compatibility.
Key Spark 4-specific changes (diff against runners/spark/3/src/):
EncoderFactory — Replaced the direct ExpressionEncoder constructor
(removed in Spark 4) with BeamAgnosticEncoder, a named class
implementing both AgnosticExpressionPathEncoder (for expression
delegation via toCatalyst/fromCatalyst) and AgnosticEncoders
.StructEncoder (so Dataset.select(TypedColumn) creates an N-attribute
plan, preventing FIELD_NUMBER_MISMATCH). The toCatalyst/fromCatalyst
methods substitute the provided input expression via transformUp,
enabling correct nesting inside composite encoders like
Encoders.tuple().
EncoderHelpers — Added toExpressionEncoder() helper to handle Spark 4
built-in encoders that are AgnosticEncoder subclasses rather than
ExpressionEncoder.
GroupByKeyTranslatorBatch — Migrated from internal catalyst Expression
API (CreateNamedStruct, Literal$) to public Column API (struct(),
lit(), array()), as required by Spark 4.
BoundedDatasetFactory — Use classic.Dataset$.MODULE$.ofRows() as
Dataset moved to org.apache.spark.sql.classic in Spark 4.
ScalaInterop — Replace WrappedArray.ofRef (removed in Scala 2.13)
with JavaConverters.asScalaBuffer().toList() in seqOf().
GroupByKeyHelpers, CombinePerKeyTranslatorBatch — Replace
TraversableOnce with IterableOnce (Scala 2.13 rename).
SparkStructuredStreamingPipelineResult — Replace sparkproject.guava
with Beam's vendored Guava.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* ci: add Spark 4 PreCommit and PostCommit workflows
Add GitHub Actions workflows for the Spark 4 runner module:
- beam_PreCommit_Java_Spark4_Versions: runs sparkVersionsTest on
changes to runners/spark/**. Currently a no-op (the sparkVersions
map is empty) but scaffolds future patch version coverage.
- beam_PostCommit_Java_ValidatesRunner_Spark4StructuredStreaming: runs
the structured streaming test suite on Java 17.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* Add PreCommit Java Spark4 Versions workflow
* Add cancellation support to Spark pipeline execution
* Remove unused endOfData() call in close method
Remove endOfData() call in close method.
* build: add Spark 4 job-server and container modules
Add job-server and container build configurations for Spark 4,
mirroring the existing Spark 3 job-server setup. The container
uses eclipse-temurin:17 (Spark 4 requires Java 17). The shared
spark_job_server.gradle gains a requireJavaVersion conditional
for Spark 4 parent projects.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* build: remove spark.driver.host workaround from Spark 4 build
The hostname binding hack is no longer needed now that the local
machine resolves its hostname to 127.0.0.1 via /etc/hosts.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* docs: add Spark 4 runner entry to CHANGES.md
Called out in /ultrareview as a missing contributor checklist item.
Adds a Highlight line and a New Features / Improvements entry under
the 2.74.0 Unreleased section, referencing issue #36841.
* docs: explain classic.SparkSession downcast in BoundedDatasetFactory
Per /ultrareview feedback: the one-line comment didn't make clear why
the cast is safe. Expand it to note that SparkSession.builder() always
returns a classic.SparkSession at runtime, which is why the downcast
avoids reflection.
* fix: log warning when neither WrappedArray nor ArraySeq class is found
Per /ultrareview feedback: the fallback branch silently swallowed the
second ClassNotFoundException. In practice one of the two classes is
always present (Scala 2.12 vs 2.13 stdlib), but a silent skip could
mask a broken classpath. Emit a LOG.warn instead.
* build: compare spark_version numerically via isSparkAtLeast helper
Per /ultrareview feedback: the five `"$spark_version" >= "3.5.0"` checks
were lexicographic string comparisons. They happened to work for 3.5.0
and 4.0.2 only because '4' > '3' as chars — a future "3.10.0" release
would compare less than "3.5.0" and silently drop the Spark 3.5+
dependencies and exclusions.
Introduce an `isSparkAtLeast` closure that tokenizes on `.` and `-`,
keeps numeric parts, and compares component-by-component. Replace all
five call sites.
* [Spark Runner] Slim Spark 4 to override-only files
With spark_runner.gradle now layering per-major source overrides on top
of the shared base, runners/spark/4/src/ no longer needs to duplicate
62 byte-identical structured-streaming files. Keep only the 11 files
that actually differ for Spark 4 / Scala 2.13. Switch the build.gradle
to spark_major = '4' (the new mechanism) and bump spark_versions to 3,4.
Compiled output unchanged — the deleted files are reproduced identically
inside build/source-overrides by the Copy task.
* [Spark Runner] Use java.io.Serializable in DoFnRunnerFactory base
scala.Serializable was removed in Scala 2.13. java.io.Serializable
works identically on both Scala 2.12 and 2.13, so this can live in
the shared base instead of needing a Spark-4-only override file.
* [Spark Runner] Null-guard error message logging in EvaluationContext base
Wrap Throwables.getRootCause(e).getMessage() in String.valueOf(...)
to make the error logging robust to a null root-cause message. The
behaviour change applies equally to Spark 3 and Spark 4, so the
fix lives in the shared base and the Spark-4 override is dropped.
* [Spark Runner] Cancel execution future and use Beam-vendored Guava in PipelineResult
Two changes that previously lived only in the Spark-4 override and
are equally valid for Spark 3:
1. cancel() now actually cancels the executing future
(pipelineExecution.cancel(true)) in addition to setting the state
to CANCELLED. Without this, calling cancel() left the pipeline
running silently — a real bug, not a Spark-4 specific concern.
2. Switch from Spark's shaded guava (org.sparkproject.guava) to the
Beam-vendored guava that is already on the classpath. Spark 4
no longer exposes the sparkproject guava package; using the
vendored one removes the version coupling for both runners.
* ci: re-trigger to clear flaky UnboundedScheduledExecutorServiceTest
Empty commit to re-run CI. The only failure on the prior head was
UnboundedScheduledExecutorServiceTest.testThreadsAreAddedOnlyAsNeededWithContention,
a known flake (#31590) — the test itself acknowledges
contention-induced extra threads in its inline comment. Squash or
drop on rebase before merge.
* [Spark Runner] Fix maxTimestamp to handle multi-window values
Iterables.getOnlyElement(windows) crashes with IllegalArgumentException
when a WindowedValue is associated with more than one window (e.g. after
a sliding window assignment). Compute the max maxTimestamp() across all
associated windows instead, falling back to a clear error if the iterable
is unexpectedly empty.
Applied identically to the shared base and the Spark 4 override. Flagged
by Gemini Code Assist on PR #38255.
* [Spark Runner] Drop unchecked cast in BoundedDatasetFactory.split
source.split returns List<? extends BoundedSource<T>>, which already
satisfies the subsequent stream usage. The cast was unchecked and would
trip heap-pollution warnings. Applied identically to the shared base
and the Spark 4 override. Flagged by Gemini Code Assist on PR #38255.
* [Spark Runner] Drop redundant Iterator cast in Spark 4 GroupByKeyTranslatorBatch
The (Iterator<V>) cast inside fun2 is redundant: fun2's signature
infers the iterator type. The shared base translator at the analogous
call site already calls iterableOnce(it) without a cast. Flagged by
Gemini Code Assist on PR #38255.
* [Spark Runner] Spark 4 EncoderFactory: stable constructor lookup + document trait setter
Replace getConstructors()[0] (JVM-defined ordering, not stable) with a
helper that picks the widest public constructor. The downstream switch
already dispatches on parameter count to pick the right argument shape
per Spark version, so this just makes the choice deterministic.
Also document the org$apache$spark...$_setter_$isStruct_$eq method —
it is the synthetic setter the Scala compiler emits for trait val fields,
required when implementing AgnosticEncoders.StructEncoder from Java.
Both flagged by Gemini Code Assist on PR #38255.
* [Spark Runner] Fix Javadoc/comment typos flagged by Gemini
Three trivial typos flagged on PR #38255 round 2 review, applied
identically to the shared base and the Spark 4 override:
- CombinePerKeyTranslatorBatch: "other there other missing features?"
-> "are there other missing features?"
- GroupByKeyTranslatorBatch: "build-in" -> "built-in"
- EncoderHelpers: PRIMITIV_TYPES -> PRIMITIVE_TYPES (constant + caller)
* [Spark Runner] Switch EncoderFactory.invoke on the right constructor
In EncoderFactory.invoke(Expression obj, ...), the switch was keyed on
STATIC_INVOKE_CONSTRUCTOR.getParameterCount() but the body actually
calls INVOKE_CONSTRUCTOR. This worked by coincidence: across the
supported Spark 3.x versions both constructors happen to share the
same parameter counts at the same dispatch points. A future Spark
release where the two diverge would silently pick the wrong branch.
Switch on INVOKE_CONSTRUCTOR.getParameterCount() to match the
constructor that is actually invoked, and align with the convention
used by newInstance() further down. In the Spark 4 override this also
lets us collapse the `case 8: case 9:` fallthrough back to a single
`case 8:`, since INVOKE_CONSTRUCTOR remains 8 params in Spark 4 even
though STATIC_INVOKE_CONSTRUCTOR grew to 9.
Applied identically to the shared base and the Spark 4 override.
Flagged by Gemini Code Assist on PR #38255.
* Update CHANGES.md
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
* [Spark 4] Drop redundant Collection cast in GroupByKeyHelpers
WindowedValue#getWindows() returns Collection<? extends BoundedWindow>,
which is already an Iterable and can be passed straight to
ScalaInterop.scalaIterator(...). The intermediate local variable and the
unchecked cast to Collection<BoundedWindow> were redundant.
Applied in both the shared base and the Spark 4 override.
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
* [Spark 4] Add module README with slf4j-jdk14 known-issue note
Documents the Spark 4 runner's requirements (Java 17, Scala 2.13,
Spark 4.0.x, batch-only) and the slf4j-jdk14 ↔ jul-to-slf4j conflict
that is the Spark 4 manifestation of #26985 (fixed for Spark 3 in
#27001). The shared spark_runner.gradle already excludes slf4j-jdk14
for in-tree builds; this note tells downstream consumers to mirror the
exclude when assembling their own runtime classpath against
beam-runners-spark-4.
* [runners-spark] Address Gemini nits: use encoder terminology in exception messages
* Trigger Build
* ci: re-trigger to clear flaky FlinkRequiresStableInputTest
The PreCommit Java failure on the previous run was a single timeout in
FlinkRequiresStableInputTest.testParDoRequiresStableInputPortable
(:runners:flink:1.17:test) — known flake tracked in #21333. This PR
does not touch any Flink code. Squash or drop on rebase before merge.
* ci: re-trigger to clear flaky SqsIOWriteBatchesTest.testWriteBatchesToDynamicWithStrictTimeout
Wall-clock-timing test (100ms inter-message + 150ms strict batch
timeout) in sdks/java/io/amazon-web-services2 SQS — unrelated to
this PR (no AWS2/SQS/Direct-runner files touched), and master is
green for the same PreCommit on 6106b30.
* ci: re-trigger to clear Maven Central 403 on Windows wordcount
`Java Wordcount Direct Runner (windows-latest)` failed at the
:buildSrc configure step with HTTP 403 fetching legacy Spotless
5.6.1 transitive deps from repo.maven.apache.org
(spotless-lib:2.7.0, durian-*:1.2.0, jgit:5.8.0). Network/infra
flake — PR doesn't touch examples or buildSrc, master 'Java Tests'
workflow consistently green.
* ci: re-trigger to clear flaky Spotless + GCP IO Direct PreCommits
Both checks failed on the prior empty retry commit (e19b80c).
Reproduced locally at e19b80c: spotlessCheck and Spark
checkStyleMain/Test all pass. PR doesn't touch any GCP IO code,
and both checks were green on the immediately preceding branch
commits (5abbb21, 604037f) and on master (6106b30, e01f711).
Treating as infra flakes; squash before merge.
* [runners-spark] Cover Scala-array fallback and EvaluationContext error paths
Address codecov/patch on PR #38255 by exercising the new branches added for
Scala 2.13 / null-safe error logging:
- Refactor SparkRunnerKryoRegistrator's nested Scala-array Class.forName
fallback into a small @VisibleForTesting findFirstAvailableClass helper
and add unit tests for first-hit, fallback, no-match, and empty-input
paths.
- Add EvaluationContextTest covering the catch (RuntimeException) /
catch (Exception) blocks in evaluate() and collect(), including the
null-message path that motivated the String.valueOf wrap.
* [runners-spark] spotless: inline two findFirstAvailableClass calls in test
* flaky SqsIOWriteBatchesTest retry
* flaky ExampleEchoPipelineTest retry
* rebase cleanup: drop duplicate isSparkAtLeast helper now in master via #38324
* Address @Abacn 2026-05-07 review
- SparkRunnerKryoRegistrator: throw IllegalStateException instead of
LOG.warn when neither ArraySeq$ofRef (Scala 2.13) nor WrappedArray$ofRef
(Scala 2.12) is on the classpath, so the missing class isn't silently
ignored. Drops the now-unused Logger field and slf4j imports.
- spark_runner.gradle: declare org.apache.spark:spark-connect-shims_2.13
as a provided dep gated on isSparkAtLeast("4.0.0"). Spark 4 splits the
Connect shim classes out of spark-sql; with enableStrictDependencies
this surfaced as analyzeClassesDependencies usedUndeclaredArtifacts.
The artifact does not exist for Spark 3, so the gate prevents Spark 3
resolution failures.
- runners/spark/4/build.gradle: drop the empty sparkVersions test
scaffolding (no additional Spark 4.x patch versions to test against
yet) and delete the now-unused
.github/workflows/beam_PreCommit_Java_Spark4_Versions.yml workflow
+ its README.md row.
- EncoderFactory (shared base): revert the line 94 switch to
STATIC_INVOKE_CONSTRUCTOR.getParameterCount(), keeping Spark 3 behavior
byte-for-byte unchanged. Spark 4's complete EncoderFactory override
under runners/spark/4/src/.../EncoderFactory.java is unaffected.
- CHANGES.md: drop the Highlights line for Spark 4. Will re-add when
ValidatesRunner tests are set up and confirmed working, matching the
Phase 1 #38324 pattern.
- runners/spark/4/job-server/container: delete the entire module
(build.gradle + Dockerfile) and remove its include() from
settings.gradle.kts. Per @Abacn's offer to defer the container module
to portable runner support later. The fat-jar :runners:spark:4:job-server
module is kept.
---------
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
aIbrahiim
pushed a commit
to aIbrahiim/beam
that referenced
this pull request
May 12, 2026
* build: add Spark 4.0.2 version property and Scala 2.13 support
Add spark4_version (4.0.2) to BeamModulePlugin alongside the existing
spark3_version. Update spark_runner.gradle to conditionally select the
correct Scala library (2.13 vs 2.12), Jackson module, Kafka test
dependency, and require Java 17 when building against Spark 4.
Register the new :runners:spark:4 module in settings.gradle.kts.
These changes are purely additive — all conditionals gate on
spark_version.startsWith("4") or spark_scala_version == '2.13', leaving
the Spark 3 build path untouched.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* refactor: make shared Spark source compatible with Scala 2.12 and 2.13
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* build: add runners/spark/4/ build configuration
Add the Gradle build file for the Spark 4 structured streaming runner.
The module mirrors runners/spark/3/ — it inherits the shared RDD-base
source from runners/spark/src/ via copySourceBase and adds its own
Structured Streaming implementation in src/main/java.
Key differences from the Spark 3 build:
- Uses spark4_version (4.0.2) with Scala 2.13.
- Excludes DStream-based streaming tests (Spark 4 supports only
structured streaming batch).
- Unconditionally adds --add-opens JVM flags required by Kryo on
Java 17 (Spark 4's minimum).
- Binds Spark driver to 127.0.0.1 for macOS compatibility.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* feat: add Spark 4 structured streaming runner source
Add the Spark 4 structured streaming runner implementation and tests.
Most files are adapted from the Spark 3 structured streaming runner
with targeted changes for Spark 4 / Scala 2.13 API compatibility.
Key Spark 4-specific changes (diff against runners/spark/3/src/):
EncoderFactory — Replaced the direct ExpressionEncoder constructor
(removed in Spark 4) with BeamAgnosticEncoder, a named class
implementing both AgnosticExpressionPathEncoder (for expression
delegation via toCatalyst/fromCatalyst) and AgnosticEncoders
.StructEncoder (so Dataset.select(TypedColumn) creates an N-attribute
plan, preventing FIELD_NUMBER_MISMATCH). The toCatalyst/fromCatalyst
methods substitute the provided input expression via transformUp,
enabling correct nesting inside composite encoders like
Encoders.tuple().
EncoderHelpers — Added toExpressionEncoder() helper to handle Spark 4
built-in encoders that are AgnosticEncoder subclasses rather than
ExpressionEncoder.
GroupByKeyTranslatorBatch — Migrated from internal catalyst Expression
API (CreateNamedStruct, Literal$) to public Column API (struct(),
lit(), array()), as required by Spark 4.
BoundedDatasetFactory — Use classic.Dataset$.MODULE$.ofRows() as
Dataset moved to org.apache.spark.sql.classic in Spark 4.
ScalaInterop — Replace WrappedArray.ofRef (removed in Scala 2.13)
with JavaConverters.asScalaBuffer().toList() in seqOf().
GroupByKeyHelpers, CombinePerKeyTranslatorBatch — Replace
TraversableOnce with IterableOnce (Scala 2.13 rename).
SparkStructuredStreamingPipelineResult — Replace sparkproject.guava
with Beam's vendored Guava.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* ci: add Spark 4 PreCommit and PostCommit workflows
Add GitHub Actions workflows for the Spark 4 runner module:
- beam_PreCommit_Java_Spark4_Versions: runs sparkVersionsTest on
changes to runners/spark/**. Currently a no-op (the sparkVersions
map is empty) but scaffolds future patch version coverage.
- beam_PostCommit_Java_ValidatesRunner_Spark4StructuredStreaming: runs
the structured streaming test suite on Java 17.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* Add PreCommit Java Spark4 Versions workflow
* Add cancellation support to Spark pipeline execution
* Remove unused endOfData() call in close method
Remove endOfData() call in close method.
* build: add Spark 4 job-server and container modules
Add job-server and container build configurations for Spark 4,
mirroring the existing Spark 3 job-server setup. The container
uses eclipse-temurin:17 (Spark 4 requires Java 17). The shared
spark_job_server.gradle gains a requireJavaVersion conditional
for Spark 4 parent projects.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* build: remove spark.driver.host workaround from Spark 4 build
The hostname binding hack is no longer needed now that the local
machine resolves its hostname to 127.0.0.1 via /etc/hosts.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* docs: add Spark 4 runner entry to CHANGES.md
Called out in /ultrareview as a missing contributor checklist item.
Adds a Highlight line and a New Features / Improvements entry under
the 2.74.0 Unreleased section, referencing issue apache#36841.
* docs: explain classic.SparkSession downcast in BoundedDatasetFactory
Per /ultrareview feedback: the one-line comment didn't make clear why
the cast is safe. Expand it to note that SparkSession.builder() always
returns a classic.SparkSession at runtime, which is why the downcast
avoids reflection.
* fix: log warning when neither WrappedArray nor ArraySeq class is found
Per /ultrareview feedback: the fallback branch silently swallowed the
second ClassNotFoundException. In practice one of the two classes is
always present (Scala 2.12 vs 2.13 stdlib), but a silent skip could
mask a broken classpath. Emit a LOG.warn instead.
* build: compare spark_version numerically via isSparkAtLeast helper
Per /ultrareview feedback: the five `"$spark_version" >= "3.5.0"` checks
were lexicographic string comparisons. They happened to work for 3.5.0
and 4.0.2 only because '4' > '3' as chars — a future "3.10.0" release
would compare less than "3.5.0" and silently drop the Spark 3.5+
dependencies and exclusions.
Introduce an `isSparkAtLeast` closure that tokenizes on `.` and `-`,
keeps numeric parts, and compares component-by-component. Replace all
five call sites.
* [Spark Runner] Slim Spark 4 to override-only files
With spark_runner.gradle now layering per-major source overrides on top
of the shared base, runners/spark/4/src/ no longer needs to duplicate
62 byte-identical structured-streaming files. Keep only the 11 files
that actually differ for Spark 4 / Scala 2.13. Switch the build.gradle
to spark_major = '4' (the new mechanism) and bump spark_versions to 3,4.
Compiled output unchanged — the deleted files are reproduced identically
inside build/source-overrides by the Copy task.
* [Spark Runner] Use java.io.Serializable in DoFnRunnerFactory base
scala.Serializable was removed in Scala 2.13. java.io.Serializable
works identically on both Scala 2.12 and 2.13, so this can live in
the shared base instead of needing a Spark-4-only override file.
* [Spark Runner] Null-guard error message logging in EvaluationContext base
Wrap Throwables.getRootCause(e).getMessage() in String.valueOf(...)
to make the error logging robust to a null root-cause message. The
behaviour change applies equally to Spark 3 and Spark 4, so the
fix lives in the shared base and the Spark-4 override is dropped.
* [Spark Runner] Cancel execution future and use Beam-vendored Guava in PipelineResult
Two changes that previously lived only in the Spark-4 override and
are equally valid for Spark 3:
1. cancel() now actually cancels the executing future
(pipelineExecution.cancel(true)) in addition to setting the state
to CANCELLED. Without this, calling cancel() left the pipeline
running silently — a real bug, not a Spark-4 specific concern.
2. Switch from Spark's shaded guava (org.sparkproject.guava) to the
Beam-vendored guava that is already on the classpath. Spark 4
no longer exposes the sparkproject guava package; using the
vendored one removes the version coupling for both runners.
* ci: re-trigger to clear flaky UnboundedScheduledExecutorServiceTest
Empty commit to re-run CI. The only failure on the prior head was
UnboundedScheduledExecutorServiceTest.testThreadsAreAddedOnlyAsNeededWithContention,
a known flake (apache#31590) — the test itself acknowledges
contention-induced extra threads in its inline comment. Squash or
drop on rebase before merge.
* [Spark Runner] Fix maxTimestamp to handle multi-window values
Iterables.getOnlyElement(windows) crashes with IllegalArgumentException
when a WindowedValue is associated with more than one window (e.g. after
a sliding window assignment). Compute the max maxTimestamp() across all
associated windows instead, falling back to a clear error if the iterable
is unexpectedly empty.
Applied identically to the shared base and the Spark 4 override. Flagged
by Gemini Code Assist on PR apache#38255.
* [Spark Runner] Drop unchecked cast in BoundedDatasetFactory.split
source.split returns List<? extends BoundedSource<T>>, which already
satisfies the subsequent stream usage. The cast was unchecked and would
trip heap-pollution warnings. Applied identically to the shared base
and the Spark 4 override. Flagged by Gemini Code Assist on PR apache#38255.
* [Spark Runner] Drop redundant Iterator cast in Spark 4 GroupByKeyTranslatorBatch
The (Iterator<V>) cast inside fun2 is redundant: fun2's signature
infers the iterator type. The shared base translator at the analogous
call site already calls iterableOnce(it) without a cast. Flagged by
Gemini Code Assist on PR apache#38255.
* [Spark Runner] Spark 4 EncoderFactory: stable constructor lookup + document trait setter
Replace getConstructors()[0] (JVM-defined ordering, not stable) with a
helper that picks the widest public constructor. The downstream switch
already dispatches on parameter count to pick the right argument shape
per Spark version, so this just makes the choice deterministic.
Also document the org$apache$spark...$_setter_$isStruct_$eq method —
it is the synthetic setter the Scala compiler emits for trait val fields,
required when implementing AgnosticEncoders.StructEncoder from Java.
Both flagged by Gemini Code Assist on PR apache#38255.
* [Spark Runner] Fix Javadoc/comment typos flagged by Gemini
Three trivial typos flagged on PR apache#38255 round 2 review, applied
identically to the shared base and the Spark 4 override:
- CombinePerKeyTranslatorBatch: "other there other missing features?"
-> "are there other missing features?"
- GroupByKeyTranslatorBatch: "build-in" -> "built-in"
- EncoderHelpers: PRIMITIV_TYPES -> PRIMITIVE_TYPES (constant + caller)
* [Spark Runner] Switch EncoderFactory.invoke on the right constructor
In EncoderFactory.invoke(Expression obj, ...), the switch was keyed on
STATIC_INVOKE_CONSTRUCTOR.getParameterCount() but the body actually
calls INVOKE_CONSTRUCTOR. This worked by coincidence: across the
supported Spark 3.x versions both constructors happen to share the
same parameter counts at the same dispatch points. A future Spark
release where the two diverge would silently pick the wrong branch.
Switch on INVOKE_CONSTRUCTOR.getParameterCount() to match the
constructor that is actually invoked, and align with the convention
used by newInstance() further down. In the Spark 4 override this also
lets us collapse the `case 8: case 9:` fallthrough back to a single
`case 8:`, since INVOKE_CONSTRUCTOR remains 8 params in Spark 4 even
though STATIC_INVOKE_CONSTRUCTOR grew to 9.
Applied identically to the shared base and the Spark 4 override.
Flagged by Gemini Code Assist on PR apache#38255.
* Update CHANGES.md
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
* [Spark 4] Drop redundant Collection cast in GroupByKeyHelpers
WindowedValue#getWindows() returns Collection<? extends BoundedWindow>,
which is already an Iterable and can be passed straight to
ScalaInterop.scalaIterator(...). The intermediate local variable and the
unchecked cast to Collection<BoundedWindow> were redundant.
Applied in both the shared base and the Spark 4 override.
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
* [Spark 4] Add module README with slf4j-jdk14 known-issue note
Documents the Spark 4 runner's requirements (Java 17, Scala 2.13,
Spark 4.0.x, batch-only) and the slf4j-jdk14 ↔ jul-to-slf4j conflict
that is the Spark 4 manifestation of apache#26985 (fixed for Spark 3 in
apache#27001). The shared spark_runner.gradle already excludes slf4j-jdk14
for in-tree builds; this note tells downstream consumers to mirror the
exclude when assembling their own runtime classpath against
beam-runners-spark-4.
* [runners-spark] Address Gemini nits: use encoder terminology in exception messages
* Trigger Build
* ci: re-trigger to clear flaky FlinkRequiresStableInputTest
The PreCommit Java failure on the previous run was a single timeout in
FlinkRequiresStableInputTest.testParDoRequiresStableInputPortable
(:runners:flink:1.17:test) — known flake tracked in apache#21333. This PR
does not touch any Flink code. Squash or drop on rebase before merge.
* ci: re-trigger to clear flaky SqsIOWriteBatchesTest.testWriteBatchesToDynamicWithStrictTimeout
Wall-clock-timing test (100ms inter-message + 150ms strict batch
timeout) in sdks/java/io/amazon-web-services2 SQS — unrelated to
this PR (no AWS2/SQS/Direct-runner files touched), and master is
green for the same PreCommit on 6106b30.
* ci: re-trigger to clear Maven Central 403 on Windows wordcount
`Java Wordcount Direct Runner (windows-latest)` failed at the
:buildSrc configure step with HTTP 403 fetching legacy Spotless
5.6.1 transitive deps from repo.maven.apache.org
(spotless-lib:2.7.0, durian-*:1.2.0, jgit:5.8.0). Network/infra
flake — PR doesn't touch examples or buildSrc, master 'Java Tests'
workflow consistently green.
* ci: re-trigger to clear flaky Spotless + GCP IO Direct PreCommits
Both checks failed on the prior empty retry commit (e19b80c).
Reproduced locally at e19b80c: spotlessCheck and Spark
checkStyleMain/Test all pass. PR doesn't touch any GCP IO code,
and both checks were green on the immediately preceding branch
commits (5abbb21, 604037f) and on master (6106b30, e01f711).
Treating as infra flakes; squash before merge.
* [runners-spark] Cover Scala-array fallback and EvaluationContext error paths
Address codecov/patch on PR apache#38255 by exercising the new branches added for
Scala 2.13 / null-safe error logging:
- Refactor SparkRunnerKryoRegistrator's nested Scala-array Class.forName
fallback into a small @VisibleForTesting findFirstAvailableClass helper
and add unit tests for first-hit, fallback, no-match, and empty-input
paths.
- Add EvaluationContextTest covering the catch (RuntimeException) /
catch (Exception) blocks in evaluate() and collect(), including the
null-message path that motivated the String.valueOf wrap.
* [runners-spark] spotless: inline two findFirstAvailableClass calls in test
* flaky SqsIOWriteBatchesTest retry
* flaky ExampleEchoPipelineTest retry
* rebase cleanup: drop duplicate isSparkAtLeast helper now in master via apache#38324
* Address @Abacn 2026-05-07 review
- SparkRunnerKryoRegistrator: throw IllegalStateException instead of
LOG.warn when neither ArraySeq$ofRef (Scala 2.13) nor WrappedArray$ofRef
(Scala 2.12) is on the classpath, so the missing class isn't silently
ignored. Drops the now-unused Logger field and slf4j imports.
- spark_runner.gradle: declare org.apache.spark:spark-connect-shims_2.13
as a provided dep gated on isSparkAtLeast("4.0.0"). Spark 4 splits the
Connect shim classes out of spark-sql; with enableStrictDependencies
this surfaced as analyzeClassesDependencies usedUndeclaredArtifacts.
The artifact does not exist for Spark 3, so the gate prevents Spark 3
resolution failures.
- runners/spark/4/build.gradle: drop the empty sparkVersions test
scaffolding (no additional Spark 4.x patch versions to test against
yet) and delete the now-unused
.github/workflows/beam_PreCommit_Java_Spark4_Versions.yml workflow
+ its README.md row.
- EncoderFactory (shared base): revert the line 94 switch to
STATIC_INVOKE_CONSTRUCTOR.getParameterCount(), keeping Spark 3 behavior
byte-for-byte unchanged. Spark 4's complete EncoderFactory override
under runners/spark/4/src/.../EncoderFactory.java is unaffected.
- CHANGES.md: drop the Highlights line for Spark 4. Will re-add when
ValidatesRunner tests are set up and confirmed working, matching the
Phase 1 apache#38324 pattern.
- runners/spark/4/job-server/container: delete the entire module
(build.gradle + Dockerfile) and remove its include() from
settings.gradle.kts. Per @Abacn's offer to defer the container module
to portable runner support later. The fat-jar :runners:spark:4:job-server
module is kept.
---------
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
ash6898
pushed a commit
to ash6898/beam
that referenced
this pull request
Jun 29, 2026
ash6898
pushed a commit
to ash6898/beam
that referenced
this pull request
Jun 29, 2026
* build: add Spark 4.0.2 version property and Scala 2.13 support
Add spark4_version (4.0.2) to BeamModulePlugin alongside the existing
spark3_version. Update spark_runner.gradle to conditionally select the
correct Scala library (2.13 vs 2.12), Jackson module, Kafka test
dependency, and require Java 17 when building against Spark 4.
Register the new :runners:spark:4 module in settings.gradle.kts.
These changes are purely additive — all conditionals gate on
spark_version.startsWith("4") or spark_scala_version == '2.13', leaving
the Spark 3 build path untouched.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* refactor: make shared Spark source compatible with Scala 2.12 and 2.13
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* build: add runners/spark/4/ build configuration
Add the Gradle build file for the Spark 4 structured streaming runner.
The module mirrors runners/spark/3/ — it inherits the shared RDD-base
source from runners/spark/src/ via copySourceBase and adds its own
Structured Streaming implementation in src/main/java.
Key differences from the Spark 3 build:
- Uses spark4_version (4.0.2) with Scala 2.13.
- Excludes DStream-based streaming tests (Spark 4 supports only
structured streaming batch).
- Unconditionally adds --add-opens JVM flags required by Kryo on
Java 17 (Spark 4's minimum).
- Binds Spark driver to 127.0.0.1 for macOS compatibility.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* feat: add Spark 4 structured streaming runner source
Add the Spark 4 structured streaming runner implementation and tests.
Most files are adapted from the Spark 3 structured streaming runner
with targeted changes for Spark 4 / Scala 2.13 API compatibility.
Key Spark 4-specific changes (diff against runners/spark/3/src/):
EncoderFactory — Replaced the direct ExpressionEncoder constructor
(removed in Spark 4) with BeamAgnosticEncoder, a named class
implementing both AgnosticExpressionPathEncoder (for expression
delegation via toCatalyst/fromCatalyst) and AgnosticEncoders
.StructEncoder (so Dataset.select(TypedColumn) creates an N-attribute
plan, preventing FIELD_NUMBER_MISMATCH). The toCatalyst/fromCatalyst
methods substitute the provided input expression via transformUp,
enabling correct nesting inside composite encoders like
Encoders.tuple().
EncoderHelpers — Added toExpressionEncoder() helper to handle Spark 4
built-in encoders that are AgnosticEncoder subclasses rather than
ExpressionEncoder.
GroupByKeyTranslatorBatch — Migrated from internal catalyst Expression
API (CreateNamedStruct, Literal$) to public Column API (struct(),
lit(), array()), as required by Spark 4.
BoundedDatasetFactory — Use classic.Dataset$.MODULE$.ofRows() as
Dataset moved to org.apache.spark.sql.classic in Spark 4.
ScalaInterop — Replace WrappedArray.ofRef (removed in Scala 2.13)
with JavaConverters.asScalaBuffer().toList() in seqOf().
GroupByKeyHelpers, CombinePerKeyTranslatorBatch — Replace
TraversableOnce with IterableOnce (Scala 2.13 rename).
SparkStructuredStreamingPipelineResult — Replace sparkproject.guava
with Beam's vendored Guava.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* ci: add Spark 4 PreCommit and PostCommit workflows
Add GitHub Actions workflows for the Spark 4 runner module:
- beam_PreCommit_Java_Spark4_Versions: runs sparkVersionsTest on
changes to runners/spark/**. Currently a no-op (the sparkVersions
map is empty) but scaffolds future patch version coverage.
- beam_PostCommit_Java_ValidatesRunner_Spark4StructuredStreaming: runs
the structured streaming test suite on Java 17.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* Add PreCommit Java Spark4 Versions workflow
* Add cancellation support to Spark pipeline execution
* Remove unused endOfData() call in close method
Remove endOfData() call in close method.
* build: add Spark 4 job-server and container modules
Add job-server and container build configurations for Spark 4,
mirroring the existing Spark 3 job-server setup. The container
uses eclipse-temurin:17 (Spark 4 requires Java 17). The shared
spark_job_server.gradle gains a requireJavaVersion conditional
for Spark 4 parent projects.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* build: remove spark.driver.host workaround from Spark 4 build
The hostname binding hack is no longer needed now that the local
machine resolves its hostname to 127.0.0.1 via /etc/hosts.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* docs: add Spark 4 runner entry to CHANGES.md
Called out in /ultrareview as a missing contributor checklist item.
Adds a Highlight line and a New Features / Improvements entry under
the 2.74.0 Unreleased section, referencing issue apache#36841.
* docs: explain classic.SparkSession downcast in BoundedDatasetFactory
Per /ultrareview feedback: the one-line comment didn't make clear why
the cast is safe. Expand it to note that SparkSession.builder() always
returns a classic.SparkSession at runtime, which is why the downcast
avoids reflection.
* fix: log warning when neither WrappedArray nor ArraySeq class is found
Per /ultrareview feedback: the fallback branch silently swallowed the
second ClassNotFoundException. In practice one of the two classes is
always present (Scala 2.12 vs 2.13 stdlib), but a silent skip could
mask a broken classpath. Emit a LOG.warn instead.
* build: compare spark_version numerically via isSparkAtLeast helper
Per /ultrareview feedback: the five `"$spark_version" >= "3.5.0"` checks
were lexicographic string comparisons. They happened to work for 3.5.0
and 4.0.2 only because '4' > '3' as chars — a future "3.10.0" release
would compare less than "3.5.0" and silently drop the Spark 3.5+
dependencies and exclusions.
Introduce an `isSparkAtLeast` closure that tokenizes on `.` and `-`,
keeps numeric parts, and compares component-by-component. Replace all
five call sites.
* [Spark Runner] Slim Spark 4 to override-only files
With spark_runner.gradle now layering per-major source overrides on top
of the shared base, runners/spark/4/src/ no longer needs to duplicate
62 byte-identical structured-streaming files. Keep only the 11 files
that actually differ for Spark 4 / Scala 2.13. Switch the build.gradle
to spark_major = '4' (the new mechanism) and bump spark_versions to 3,4.
Compiled output unchanged — the deleted files are reproduced identically
inside build/source-overrides by the Copy task.
* [Spark Runner] Use java.io.Serializable in DoFnRunnerFactory base
scala.Serializable was removed in Scala 2.13. java.io.Serializable
works identically on both Scala 2.12 and 2.13, so this can live in
the shared base instead of needing a Spark-4-only override file.
* [Spark Runner] Null-guard error message logging in EvaluationContext base
Wrap Throwables.getRootCause(e).getMessage() in String.valueOf(...)
to make the error logging robust to a null root-cause message. The
behaviour change applies equally to Spark 3 and Spark 4, so the
fix lives in the shared base and the Spark-4 override is dropped.
* [Spark Runner] Cancel execution future and use Beam-vendored Guava in PipelineResult
Two changes that previously lived only in the Spark-4 override and
are equally valid for Spark 3:
1. cancel() now actually cancels the executing future
(pipelineExecution.cancel(true)) in addition to setting the state
to CANCELLED. Without this, calling cancel() left the pipeline
running silently — a real bug, not a Spark-4 specific concern.
2. Switch from Spark's shaded guava (org.sparkproject.guava) to the
Beam-vendored guava that is already on the classpath. Spark 4
no longer exposes the sparkproject guava package; using the
vendored one removes the version coupling for both runners.
* ci: re-trigger to clear flaky UnboundedScheduledExecutorServiceTest
Empty commit to re-run CI. The only failure on the prior head was
UnboundedScheduledExecutorServiceTest.testThreadsAreAddedOnlyAsNeededWithContention,
a known flake (apache#31590) — the test itself acknowledges
contention-induced extra threads in its inline comment. Squash or
drop on rebase before merge.
* [Spark Runner] Fix maxTimestamp to handle multi-window values
Iterables.getOnlyElement(windows) crashes with IllegalArgumentException
when a WindowedValue is associated with more than one window (e.g. after
a sliding window assignment). Compute the max maxTimestamp() across all
associated windows instead, falling back to a clear error if the iterable
is unexpectedly empty.
Applied identically to the shared base and the Spark 4 override. Flagged
by Gemini Code Assist on PR apache#38255.
* [Spark Runner] Drop unchecked cast in BoundedDatasetFactory.split
source.split returns List<? extends BoundedSource<T>>, which already
satisfies the subsequent stream usage. The cast was unchecked and would
trip heap-pollution warnings. Applied identically to the shared base
and the Spark 4 override. Flagged by Gemini Code Assist on PR apache#38255.
* [Spark Runner] Drop redundant Iterator cast in Spark 4 GroupByKeyTranslatorBatch
The (Iterator<V>) cast inside fun2 is redundant: fun2's signature
infers the iterator type. The shared base translator at the analogous
call site already calls iterableOnce(it) without a cast. Flagged by
Gemini Code Assist on PR apache#38255.
* [Spark Runner] Spark 4 EncoderFactory: stable constructor lookup + document trait setter
Replace getConstructors()[0] (JVM-defined ordering, not stable) with a
helper that picks the widest public constructor. The downstream switch
already dispatches on parameter count to pick the right argument shape
per Spark version, so this just makes the choice deterministic.
Also document the org$apache$spark...$_setter_$isStruct_$eq method —
it is the synthetic setter the Scala compiler emits for trait val fields,
required when implementing AgnosticEncoders.StructEncoder from Java.
Both flagged by Gemini Code Assist on PR apache#38255.
* [Spark Runner] Fix Javadoc/comment typos flagged by Gemini
Three trivial typos flagged on PR apache#38255 round 2 review, applied
identically to the shared base and the Spark 4 override:
- CombinePerKeyTranslatorBatch: "other there other missing features?"
-> "are there other missing features?"
- GroupByKeyTranslatorBatch: "build-in" -> "built-in"
- EncoderHelpers: PRIMITIV_TYPES -> PRIMITIVE_TYPES (constant + caller)
* [Spark Runner] Switch EncoderFactory.invoke on the right constructor
In EncoderFactory.invoke(Expression obj, ...), the switch was keyed on
STATIC_INVOKE_CONSTRUCTOR.getParameterCount() but the body actually
calls INVOKE_CONSTRUCTOR. This worked by coincidence: across the
supported Spark 3.x versions both constructors happen to share the
same parameter counts at the same dispatch points. A future Spark
release where the two diverge would silently pick the wrong branch.
Switch on INVOKE_CONSTRUCTOR.getParameterCount() to match the
constructor that is actually invoked, and align with the convention
used by newInstance() further down. In the Spark 4 override this also
lets us collapse the `case 8: case 9:` fallthrough back to a single
`case 8:`, since INVOKE_CONSTRUCTOR remains 8 params in Spark 4 even
though STATIC_INVOKE_CONSTRUCTOR grew to 9.
Applied identically to the shared base and the Spark 4 override.
Flagged by Gemini Code Assist on PR apache#38255.
* Update CHANGES.md
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
* [Spark 4] Drop redundant Collection cast in GroupByKeyHelpers
WindowedValue#getWindows() returns Collection<? extends BoundedWindow>,
which is already an Iterable and can be passed straight to
ScalaInterop.scalaIterator(...). The intermediate local variable and the
unchecked cast to Collection<BoundedWindow> were redundant.
Applied in both the shared base and the Spark 4 override.
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
* [Spark 4] Add module README with slf4j-jdk14 known-issue note
Documents the Spark 4 runner's requirements (Java 17, Scala 2.13,
Spark 4.0.x, batch-only) and the slf4j-jdk14 ↔ jul-to-slf4j conflict
that is the Spark 4 manifestation of apache#26985 (fixed for Spark 3 in
apache#27001). The shared spark_runner.gradle already excludes slf4j-jdk14
for in-tree builds; this note tells downstream consumers to mirror the
exclude when assembling their own runtime classpath against
beam-runners-spark-4.
* [runners-spark] Address Gemini nits: use encoder terminology in exception messages
* Trigger Build
* ci: re-trigger to clear flaky FlinkRequiresStableInputTest
The PreCommit Java failure on the previous run was a single timeout in
FlinkRequiresStableInputTest.testParDoRequiresStableInputPortable
(:runners:flink:1.17:test) — known flake tracked in apache#21333. This PR
does not touch any Flink code. Squash or drop on rebase before merge.
* ci: re-trigger to clear flaky SqsIOWriteBatchesTest.testWriteBatchesToDynamicWithStrictTimeout
Wall-clock-timing test (100ms inter-message + 150ms strict batch
timeout) in sdks/java/io/amazon-web-services2 SQS — unrelated to
this PR (no AWS2/SQS/Direct-runner files touched), and master is
green for the same PreCommit on 6106b30.
* ci: re-trigger to clear Maven Central 403 on Windows wordcount
`Java Wordcount Direct Runner (windows-latest)` failed at the
:buildSrc configure step with HTTP 403 fetching legacy Spotless
5.6.1 transitive deps from repo.maven.apache.org
(spotless-lib:2.7.0, durian-*:1.2.0, jgit:5.8.0). Network/infra
flake — PR doesn't touch examples or buildSrc, master 'Java Tests'
workflow consistently green.
* ci: re-trigger to clear flaky Spotless + GCP IO Direct PreCommits
Both checks failed on the prior empty retry commit (e19b80c).
Reproduced locally at e19b80c: spotlessCheck and Spark
checkStyleMain/Test all pass. PR doesn't touch any GCP IO code,
and both checks were green on the immediately preceding branch
commits (5abbb21, 604037f) and on master (6106b30, e01f711).
Treating as infra flakes; squash before merge.
* [runners-spark] Cover Scala-array fallback and EvaluationContext error paths
Address codecov/patch on PR apache#38255 by exercising the new branches added for
Scala 2.13 / null-safe error logging:
- Refactor SparkRunnerKryoRegistrator's nested Scala-array Class.forName
fallback into a small @VisibleForTesting findFirstAvailableClass helper
and add unit tests for first-hit, fallback, no-match, and empty-input
paths.
- Add EvaluationContextTest covering the catch (RuntimeException) /
catch (Exception) blocks in evaluate() and collect(), including the
null-message path that motivated the String.valueOf wrap.
* [runners-spark] spotless: inline two findFirstAvailableClass calls in test
* flaky SqsIOWriteBatchesTest retry
* flaky ExampleEchoPipelineTest retry
* rebase cleanup: drop duplicate isSparkAtLeast helper now in master via apache#38324
* Address @Abacn 2026-05-07 review
- SparkRunnerKryoRegistrator: throw IllegalStateException instead of
LOG.warn when neither ArraySeq$ofRef (Scala 2.13) nor WrappedArray$ofRef
(Scala 2.12) is on the classpath, so the missing class isn't silently
ignored. Drops the now-unused Logger field and slf4j imports.
- spark_runner.gradle: declare org.apache.spark:spark-connect-shims_2.13
as a provided dep gated on isSparkAtLeast("4.0.0"). Spark 4 splits the
Connect shim classes out of spark-sql; with enableStrictDependencies
this surfaced as analyzeClassesDependencies usedUndeclaredArtifacts.
The artifact does not exist for Spark 3, so the gate prevents Spark 3
resolution failures.
- runners/spark/4/build.gradle: drop the empty sparkVersions test
scaffolding (no additional Spark 4.x patch versions to test against
yet) and delete the now-unused
.github/workflows/beam_PreCommit_Java_Spark4_Versions.yml workflow
+ its README.md row.
- EncoderFactory (shared base): revert the line 94 switch to
STATIC_INVOKE_CONSTRUCTOR.getParameterCount(), keeping Spark 3 behavior
byte-for-byte unchanged. Spark 4's complete EncoderFactory override
under runners/spark/4/src/.../EncoderFactory.java is unaffected.
- CHANGES.md: drop the Highlights line for Spark 4. Will re-add when
ValidatesRunner tests are set up and confirmed working, matching the
Phase 1 apache#38324 pattern.
- runners/spark/4/job-server/container: delete the entire module
(build.gradle + Dockerfile) and remove its include() from
settings.gradle.kts. Per @Abacn's offer to defer the container module
to portable runner support later. The fat-jar :runners:spark:4:job-server
module is kept.
---------
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
First split-out PR from #38255 per @Abacn's review guidance
(#38255 (review)):
Three commits, one per bucket:
All hunks behaviour-identical on Spark 3.5
CHANGES.md entry added under "New Features / Improvements".
Follow-up: rebase #38255 on master once this lands