Support DataStream batch for portable flink runner and run validation runner tests #37133

Abacn · 2025-12-17T16:07:27Z

The main purpose of this PR is to start exercising validation runner tests for Flink portable runner with useDataStreamForBatch enabled. As DataSet API is removed in Flink 2, all batch workflows will use DataStream API.

Honor getUseDataStreamForBatch pipeline option for Flink portable runner
Refactor gradle scripts in preparation for Flink 2 support, including
- Flink PVR Batch PreCommit exercises useDataStreamForBatch tests, move the original Batch (based on DataSet) to a PostCommit
- Improve version-specific source overwrite logic, see below

Please add a meaningful description for your change here

Part of #36947

Fix #31631

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

gemini-code-assist · 2025-12-17T16:07:56Z

Summary of Changes

Hello @Abacn, I'm Gemini Code Assist¹! 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 represents the initial phase of introducing Apache Flink 2 support into the project. It primarily focuses on foundational build system adjustments and a critical functional update. The build changes are designed to handle Flink 2's new requirements, such as a higher Java version and changes in its core APIs, by implementing flexible source overrides and conditional dependencies. Additionally, it ensures that the Flink portable runner respects the getUseDataStreamForBatch pipeline option, providing more control over execution strategies.

Highlights

Flink 2 Build System Refactoring: Extensive refactoring of Gradle build scripts has been performed to prepare for Flink 2 support. This includes conditional Java 11 requirements for Flink 2, dynamic source directory overrides, and conditional dependency management to account for API changes like the removal of the DataSet API.
Honoring getUseDataStreamForBatch Option: The Flink portable runner now correctly honors the getUseDataStreamForBatch pipeline option, allowing users to explicitly choose the DataStream API for batch pipelines.
Docker Image Naming Conventions: Updated Docker image naming and tagging logic for Flink job server containers to accommodate Flink 2 specific conventions.
Source File Restructuring: Several Flink 1.17 specific source files have been moved to a more general location, indicating a restructuring to better support version-specific implementations and overrides for Flink 2.

Using Gemini Code Assist

The 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 /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

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 .gemini/ folder in the base of the repository. Detailed instructions can be found here.

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.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

Abacn · 2025-12-17T20:38:31Z

1 failing test:

ViewTest > testTriggeredLatestSingleton FAILED
    java.lang.RuntimeException at ViewTest.java:320

java.lang.RuntimeException: The Runner experienced the following error during execution:
java.lang.RuntimeException: Error received from SDK harness for instruction 9: java.lang.IllegalArgumentException: PCollection with more than one element accessed as a singleton view.
	at org.apache.beam.sdk.values.PCollectionViews$SingletonViewFn2.apply(PCollectionViews.java:549)
	at org.apache.beam.sdk.values.PCollectionViews$SingletonViewFn2.apply(PCollectionViews.java:475)

github-actions · 2025-12-18T00:27:54Z

Assigning reviewers:

R: @liferoad for label build.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

Abacn · 2025-12-18T15:15:49Z

R: @Amar3tto @shunping

github-actions · 2025-12-18T15:17:07Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

* Honor getUseDataStreamForBatch pipeline option for Flink portable runner * Refactor gradle scripts in preparation for Flink 2 support * Create a PostCommit run validate runner tests on legacy DataSet

Abacn · 2026-01-06T16:05:37Z

Test passed: https://github.com/apache/beam/actions/runs/20529211706/job/58977883497?pr=37133

Abacn · 2026-01-06T16:14:22Z

R: @damccorm (next release manager) @je-ik (for Flink runner) could take a look? Thanks!

damccorm

Generally this lgtm - just had a couple minor comments/questions, but they don't need to block merging for now

damccorm · 2026-01-06T21:04:14Z

runners/flink/job-server-container/flink_job_server_container.gradle

 }

 def pushContainers = project.rootProject.hasProperty(["isRelease"]) || project.rootProject.hasProperty("push-containers")
+def containerName = project.parent.name.startsWith("2") ? "flink_job_server" : "flink${project.parent.name}_job_server"


Why do we no longer need a separate container with flink 2?

Oh I see - you're moving to a tag approach. Maybe we could just do this generally for all versions? It would be nice to not need a new container each time

Sorry, this is part of #31631 (updated PR description now). Also noted there I intended to make this change in new Flink (major) version to avoid breaking existing users.

damccorm · 2026-01-06T21:06:33Z

runners/flink/flink_runner.gradle


 def overrides(versions, type, base_path) {
-  versions.collect { "${base_path}/${it}/src/${type}/java" } + ["./src/${type}/java"]
+  // order is important


Why does order matter?

For file A.java in flink/1.18/src, it is consumed by flink/1.18, flink1.19, flink1.20,...

If in Flink 1.19 it needs to be overriden again, this order makes sure that flink1.20 is consuming A.java in flink/1.19 dir

runners/flink/flink_runner.gradle

je-ik · 2026-01-07T10:11:57Z

This is great effort! 👍

github-actions bot added docker runners flink labels Dec 17, 2025

Abacn force-pushed the flink-use-datastream branch from fbb46f2 to 35d1b32 Compare December 17, 2025 16:18

Abacn marked this pull request as draft December 17, 2025 17:16

Abacn force-pushed the flink-use-datastream branch from 35d1b32 to 73c7684 Compare December 17, 2025 17:25

Abacn changed the title ~~Flink 2 support prerequisites (1/n)~~ [do not merge] Run portable flink runner validation runner test for DataStream batch Dec 17, 2025

Abacn force-pushed the flink-use-datastream branch from 73c7684 to 25a2844 Compare December 17, 2025 18:49

github-actions bot added the build label Dec 17, 2025

Abacn force-pushed the flink-use-datastream branch from c914121 to 63e5f2c Compare December 17, 2025 23:21

Abacn changed the title ~~[do not merge] Run portable flink runner validation runner test for DataStream batch~~ Support DataStream batch for portable flink runner and run validation runner tests Dec 18, 2025

Abacn marked this pull request as ready for review December 18, 2025 00:11

github-actions bot added the Next Action: Reviewers label Dec 18, 2025

Flink 2 support prerequisites

f57e4db

* Honor getUseDataStreamForBatch pipeline option for Flink portable runner * Refactor gradle scripts in preparation for Flink 2 support * Create a PostCommit run validate runner tests on legacy DataSet

Abacn force-pushed the flink-use-datastream branch from 63e5f2c to f57e4db Compare December 26, 2025 20:51

apache deleted a comment from Abacn Jan 6, 2026

damccorm approved these changes Jan 6, 2026

View reviewed changes

Abacn commented Jan 6, 2026

View reviewed changes

runners/flink/flink_runner.gradle Show resolved Hide resolved

Abacn merged commit d55cf58 into apache:master Jan 7, 2026
23 of 24 checks passed

Abacn deleted the flink-use-datastream branch January 7, 2026 01:41

Support DataStream batch for portable flink runner and run validation runner tests #37133

Support DataStream batch for portable flink runner and run validation runner tests #37133

Uh oh!

Conversation

Abacn commented Dec 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

GitHub Actions Tests Status (on master branch)

Uh oh!

gemini-code-assist bot commented Dec 17, 2025

Summary of Changes

Highlights

Footnotes

Uh oh!

Abacn commented Dec 17, 2025

Uh oh!

github-actions bot commented Dec 18, 2025

Uh oh!

Abacn commented Dec 18, 2025

Uh oh!

github-actions bot commented Dec 18, 2025

Uh oh!

Abacn commented Jan 6, 2026

Uh oh!

Abacn commented Jan 6, 2026

Uh oh!

damccorm left a comment

Choose a reason for hiding this comment

Uh oh!

damccorm Jan 6, 2026

Choose a reason for hiding this comment

Uh oh!

damccorm Jan 6, 2026

Choose a reason for hiding this comment

Uh oh!

Abacn Jan 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

damccorm Jan 6, 2026

Choose a reason for hiding this comment

Uh oh!

Abacn Jan 6, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

je-ik commented Jan 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Abacn commented Dec 17, 2025 •

edited

Loading

Abacn Jan 6, 2026 •

edited

Loading