gemini-code-assist[bot] commented on PR #37036:
URL: https://github.com/apache/beam/pull/37036#issuecomment-3629777386
## Summary of Changes
This pull request addresses a critical bug in the Flink classic runner where
pipelines utilizing 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 introducing dedicated translators and refactoring
common source translation logic, this change ensures that such pipelines,
including those using unbounded sources like KinesisIO, can now execute
successfully, significantly improving the robustness and compatibility of the
Flink runner.
### Highlights
* **Fix for Flink Classic Runner**: Resolves an `IllegalStateException` in
the Flink classic runner when `PrimitiveUnboundedRead` or
`PrimitiveBoundedRead` transforms are encountered, particularly with unbounded
sources like KinesisIO, due to missing translators.
* **New Translators Introduced**: Adds `PrimitiveUnboundedReadTranslator`
and `PrimitiveBoundedReadTranslator` to explicitly handle primitive read
transforms generated by `SplittableParDo`'s conversion logic.
* **Refactored Source Translation**: Common logic for translating unbounded
and bounded sources has been extracted into dedicated helper methods
(`translateUnboundedSource`, `translateBoundedSource`) to reduce code
duplication and improve maintainability.
* **Enhanced Test Coverage**: New unit tests have been added to verify the
correct functioning and source production of the newly introduced primitive
read translators.
<details>
<summary><b>Changelog</b></summary>
* **CHANGES.md**
* Added a bugfix entry for Flink classic runner failing with "No
translator known for PrimitiveUnboundedRead" when using unbounded source
connectors like KinesisIO after SDF-to-primitive-read conversion.
*
**runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java**
* Introduced `PrimitiveUnboundedReadTranslator<T>` and
`PrimitiveBoundedReadTranslator<T>` classes to handle primitive read transforms.
* Modified the `getTranslator()` method to prioritize lookup for
`PrimitiveUnboundedRead` and `PrimitiveBoundedRead` instances.
* Extracted common source translation logic into new private static
helper methods: `translateUnboundedSource` and `translateBoundedSource`.
* Extracted a private static helper method `getSourceParallelism` to
determine source operator parallelism.
* Refactored `UnboundedReadSourceTranslator` and
`BoundedReadSourceTranslator` to utilize the new common translation helper
methods.
*
**runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslatorsTest.java**
* Added new unit tests:
`getTranslatorReturnsPrimitiveUnboundedReadTranslator`,
`getTranslatorReturnsPrimitiveBoundedReadTranslator`,
`primitiveUnboundedReadTranslatorProducesCorrectSource`, and
`primitiveBoundedReadTranslatorProducesCorrectSource`.
* Introduced a new helper method
`applyReadSourceTransformWithTranslator` for testing purposes.
</details>
<details>
<summary><b>Activity</b></summary>
* A bot provided an initial summary of the pull request.
* The bot noted that checks were failing and would not request a review.
* The author identified a potential NullPointerException in
`StateFetchingIterators.java` affecting a test (`testSideInputWithNullDefault`)
and sought guidance on whether to fix it in this PR or a new one.
* The author issued multiple `/gemini review` commands and a `/gemini
summary` command.
* A bot review highlighted that the `PrimitiveBoundedReadTranslator` was
missing important logic (try-catch, `.returns()`, batch mode handling,
consistent `stepName`) present in `BoundedReadSourceTranslator`.
* A bot review suggested replacing the `XXXXX` placeholder in `CHANGES.md`
with the actual issue number `37035`.
* A bot review recommended extracting common logic from
`UnboundedReadSourceTranslator` and `PrimitiveUnboundedReadTranslator` into a
shared helper method.
* A bot review suggested extracting the source parallelism determination
logic into a private helper method to reduce duplication.
* A bot review suggested extracting the logic for retrieving the single
output `PCollection` into a helper method, to which the author responded,
arguing against over-abstraction for a 2-3 line duplication, and the bot
acknowledged the valid point.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
