Add primitive read translators #37036
Conversation
…ink streaming runner This PR adds explicit translators for SplittableParDo.PrimitiveUnboundedRead and SplittableParDo.PrimitiveBoundedRead to the Flink streaming runner, enabling unbounded sources like KinesisIO.read() to work with the classic Flink runner on AWS Managed Flink. Problem: - Read.Unbounded.expand() wraps sources with UnboundedSourceAsSDFWrapperFn - FlinkRunner converts SDF-based reads to PrimitiveUnboundedRead - FlinkStreamingTransformTranslators.getTranslator() had no translator for PrimitiveUnboundedRead, causing 'No translator known' errors Solution: - Add PrimitiveUnboundedReadTranslator that extracts UnboundedSource directly from transform.getSource() instead of using ReadTranslation - Add PrimitiveBoundedReadTranslator for bounded sources - Modify getTranslator() to check for these transform types explicitly Testing: - Added 4 new unit tests - All 383 Flink runner tests pass
…veBoundedRead This commit adds explicit translators for SplittableParDo.PrimitiveUnboundedRead and SplittableParDo.PrimitiveBoundedRead to the Flink streaming transform translators. The Flink classic runner calls convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary() when NOT using beam_fn_api experiment, which converts SDF-wrapped reads to PrimitiveUnboundedRead and PrimitiveBoundedRead. Without translators for these transforms, pipelines using unbounded sources like KinesisIO.read() fail with: 'No translator known for PrimitiveUnboundedRead' Changes: - Add PrimitiveUnboundedReadTranslator class - Add PrimitiveBoundedReadTranslator class - Modify getTranslator() to handle these transforms before URN lookup - Add unit tests for the new translators - Update CHANGES.md with bugfix entry Related to apache#20530
Summary of ChangesHello @dnamaz, 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 resolves a critical issue in the Flink classic runner where pipelines using certain read operations, specifically those internally converted to "PrimitiveUnboundedRead" or "PrimitiveBoundedRead" by "SplittableParDo", would fail during translation due to the absence of appropriate translators. By implementing dedicated translators for these primitive read types, the change ensures that such pipelines, including those leveraging unbounded sources like KinesisIO, can now execute successfully, thereby enhancing the robustness and compatibility of the Flink runner. Highlights
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. 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. Footnotes
|
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
assign set of reviewers There is a problem with one of the tests, it is happening in this file: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/StateFetchingIterators.java on line 302,
will throw a null pointer exception if the value is null. I believe this is causing the issue with the PR check failing in the test. testSideInputWithNullDefault, should this be fixed and expected in the test or modify the collection to use a null safe list? should it be part of this PR or create another bug? Does this currently happen in other PR checks? Am i missing something? The call chain is: |
|
/gemini review |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request correctly adds translators for PrimitiveUnboundedRead and PrimitiveBoundedRead to the Flink runner, fixing a crash when using unbounded sources without the beam_fn_api experiment. The changes are well-structured and include thorough tests.
My main feedback is regarding code duplication in the new translator classes. Both PrimitiveUnboundedReadTranslator and PrimitiveBoundedReadTranslator share a significant amount of code with their existing counterparts (UnboundedReadSourceTranslator and BoundedReadSourceTranslator). I've suggested refactoring these to use common helper methods to improve maintainability.
Additionally, PrimitiveBoundedReadTranslator appears to be missing some important logic from BoundedReadSourceTranslator for error handling and batch execution, which could lead to issues. I've provided a code suggestion to address this.
Finally, there's a minor update needed in CHANGES.md to replace the placeholder issue number.
Overall, this is a great fix, and with the suggested improvements, the code will be more robust and maintainable.
...rs/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
Show resolved
Hide resolved
...rs/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
Show resolved
Hide resolved
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
- Add translateUnboundedSource() helper method for common unbounded source translation logic - Add translateBoundedSource() helper method for common bounded source translation logic - Simplify UnboundedReadSourceTranslator to use helper method - Simplify PrimitiveUnboundedReadTranslator to use helper method - Simplify BoundedReadSourceTranslator to use helper method - Simplify PrimitiveBoundedReadTranslator to use helper method - Add missing try-catch, .returns(), and batch mode slot sharing to PrimitiveBoundedReadTranslator This refactoring reduces ~100 lines of duplicated code and ensures consistent behavior across all read translators.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request correctly addresses a failure in the Flink classic runner by adding translators for PrimitiveUnboundedRead and PrimitiveBoundedRead. The changes are well-structured, including a refactoring that extracts common source translation logic into helper methods, which improves code clarity and reduces duplication. The new translators correctly handle the primitive read transforms that are generated when SplittableParDo converts SDF-wrapped reads. The accompanying tests are thorough and validate the new functionality effectively. I have one minor suggestion to further reduce code duplication.
...rs/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
Outdated
Show resolved
Hide resolved
Extract the duplicated parallelism calculation logic from translateUnboundedSource and translateBoundedSource into a shared private helper method.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request effectively addresses a critical bug in the Flink classic runner where pipelines with unbounded sources would fail due to missing translators for PrimitiveUnboundedRead and PrimitiveBoundedRead. The solution of adding explicit translators for these types is direct and correct.
I appreciate the refactoring you've done to extract common logic for bounded and unbounded source translation into translateBoundedSource and translateUnboundedSource helper methods. This significantly improves code clarity and maintainability by reducing duplication. The accompanying unit tests are comprehensive, verifying that the correct translators are selected and that they produce the expected Flink source transformations.
I have one minor suggestion to further reduce code duplication. Overall, this is a high-quality contribution that resolves an important issue.
| PCollection<T> output = | ||
| (PCollection<T>) | ||
| Iterables.getOnlyElement(context.getCurrentTransform().getOutputs().values()); |
There was a problem hiding this comment.
This logic to retrieve the single output PCollection is duplicated in translateBoundedSource (line 273). To improve maintainability and reduce code duplication, consider extracting this into a private static helper method within this class.
For example:
@SuppressWarnings("unchecked")
private static <T> PCollection<T> getOnlyOutput(FlinkStreamingTranslationContext context) {
return (PCollection<T>)
Iterables.getOnlyElement(context.getCurrentTransform().getOutputs().values());
}You could then call getOnlyOutput(context) in both translateUnboundedSource and translateBoundedSource.
There was a problem hiding this comment.
I considered this, but felt that extracting a 2-3 line helper for code that's only duplicated in two places within the same file would be over-abstraction. The Iterables.getOnlyElement() call is already self-documenting about the expected behavior.
@gemini-code-assist
There was a problem hiding this comment.
You make a valid point. While the DRY principle is generally good for maintainability, it's also important to balance it with readability and avoiding unnecessary abstraction, especially for very short and self-documenting code snippets like Iterables.getOnlyElement(). Given that the duplication is limited to two places within the same file, your decision to keep it as is, prioritizing immediate readability and avoiding potential over-abstraction, is perfectly reasonable. Thank you for the thoughtful explanation!
|
/gemini summary |
Summary of ChangesThis pull request addresses a critical bug in the Flink classic runner where pipelines utilizing certain read operations, specifically those internally converted to Highlights
Changelog
Activity
|
|
assign set of reviewers |
|
Assigning reviewers: R: @shunping added as fallback since no labels match configuration 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). |
|
Reminder, please take a look at this pr: @shunping |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @shunping added as fallback since no labels match configuration Available commands:
|
|
Reminder, please take a look at this pr: @shunping |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
2 participants
Pull Request: Add PrimitiveUnboundedRead/PrimitiveBoundedRead Translators to Flink Runner
Title
[Flink Runner] Add translators for PrimitiveUnboundedRead and PrimitiveBoundedReadDescription
What is this PR doing?
This PR adds explicit translators for
SplittableParDo.PrimitiveUnboundedReadandSplittableParDo.PrimitiveBoundedReadto the Flink streaming transform translators. These translators handle the case whereRead.UnboundedandRead.Boundedare converted to primitive reads bySplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary().Why is this needed?
The Flink classic runner (non-portable) calls
convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary()when NOT using thebeam_fn_apiexperiment. This converts SDF-wrapped reads toPrimitiveUnboundedReadandPrimitiveBoundedReadtransforms. However, there were no registered translators for these transforms, causing pipelines using unbounded sources (likeKinesisIO.read()) to fail with:How does it work?
The new translators:
PrimitiveUnboundedReadTranslator: Extracts theUnboundedSourcefromPrimitiveUnboundedRead.getSource()and creates aFlinkUnboundedSource, following the same pattern as the existingUnboundedReadSourceTranslator.PrimitiveBoundedReadTranslator: Extracts theBoundedSourcefromPrimitiveBoundedRead.getSource()and creates aFlinkBoundedSource, following the same pattern as the existingBoundedReadSourceTranslator.The key difference from the existing translators is that they retrieve the source directly from the transform (
transform.getSource()) rather than usingReadTranslation.unboundedSourceFromTransform(), sincePrimitiveUnboundedReadandPrimitiveBoundedReadare not standard Read transforms with URNs.Changes
FlinkStreamingTransformTranslators.java:PrimitiveUnboundedReadTranslator<T>classPrimitiveBoundedReadTranslator<T>classgetTranslator()to check forPrimitiveUnboundedReadandPrimitiveBoundedReadinstances before URN lookupFlinkStreamingTransformTranslatorsTest.java:getTranslatorReturnsPrimitiveUnboundedReadTranslator()testgetTranslatorReturnsPrimitiveBoundedReadTranslator()testprimitiveUnboundedReadTranslatorProducesCorrectSource()testprimitiveBoundedReadTranslatorProducesCorrectSource()testCHANGES.md:Issue
Fixes #37035
Created issue: #37035
Related to #20530 (Use SDF read as default)
Checklist
./gradlew :runners:flink:1.18:spotlessApplyFlinkStreamingTransformTranslatorsTest.java./gradlew :runners:flink:1.18:test)CHANGES.mdupdated and formatted with./gradlew formatChangesTesting
Unit Tests
./gradlew :runners:flink:1.18:test # BUILD SUCCESSFUL - all tests passIntegration Testing
Tested on AWS Managed Apache Flink with a real pipeline using
KinesisIO.read():Test environment:
Backwards Compatibility
This change is fully backwards compatible:
PrimitiveUnboundedReadorPrimitiveBoundedReadtransforms are present (which previously caused failures)Read.BoundedandRead.Unboundedwith URNs continue to use the existingReadSourceTranslatorPerformance
No performance impact expected. The new translators use the same
FlinkUnboundedSourceandFlinkBoundedSourceimplementations as the existing translators.