gemini-code-assist[bot] commented on PR #37762: URL: https://github.com/apache/beam/pull/37762#issuecomment-3996759594
## Summary of Changes Hello, I'm Gemini Code Assist[^1]! 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 enhances the robustness of Dataflow streaming pipelines by introducing a new configuration option that allows workers to gracefully handle and skip input elements that cannot be successfully decoded. This prevents pipeline failures caused by corrupted data, ensuring continuous processing by logging errors and moving past malformed inputs. ### Highlights * **New Pipeline Option**: Introduced a new Dataflow streaming pipeline option, `skipInputElementsWithDecodingExceptions`, which allows the pipeline to log and skip input elements that fail to decode successfully from the streaming backend. This option defaults to off but can be enabled during pipeline updates. * **Error Handling in Readers**: Modified `PubsubReader`, `UngroupedWindmillReader`, and `WindmillReaderIteratorBase` to incorporate the new `skipDecodingErrors` flag. When enabled, these readers will catch `IOException` during decoding, log the error, and skip the problematic element instead of throwing an exception and potentially halting the pipeline. * **Keyed Work Item Processing**: Updated `WindmillKeyedWorkItem` to accept a `skipExceptions` flag. The element parsing logic was refactored into a dedicated `parseElem` method, which now logs decoding errors and returns null for corrupted elements if `skipExceptions` is true, allowing the iterable to filter them out. * **Improved Robustness for PaneInfo**: Added validation to `PaneInfo.Encoding.fromTag` to throw a `CoderException` for invalid pane encoding tags, enhancing error detection for corrupted metadata. * **Testing and Suppressions**: Added comprehensive unit tests in `StreamingDataflowWorkerTest`, `WindmillKeyedWorkItemTest`, and `WindmillReaderIteratorBaseTest` to verify the functionality of skipping corrupted input elements. Several `@SuppressWarnings` annotations were also removed or updated, and `@Internal` was added to `WindmillKeyedWorkItem`. 🧠**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](https://codeassist.google/code-review/login) to enable Memory in your admin console.** <details> <summary><b>Changelog</b></summary> * **runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java** * Added `getSkipInputElementsWithDecodingExceptions()` and `setSkipInputElementsWithDecodingExceptions()` methods to introduce a new pipeline option. * **runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubReader.java** * Imported `DataflowStreamingPipelineOptions`. * Added `skipDecodingErrors` field and updated constructors to accept this new boolean flag. * Modified the `Factory` class to retrieve the `skipInputElementsWithDecodingExceptions` option from pipeline options and pass it to the `PubsubReader` constructor. * Updated the `PubsubReaderIterator` constructor to pass the `skipDecodingErrors` flag to its superclass. * **runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/UngroupedWindmillReader.java** * Imported `DataflowStreamingPipelineOptions` and `checkNotNull`. * Added `skipDecodingErrors` field and updated constructors to accept this new boolean flag. * Modified the `Factory` class to retrieve the `skipInputElementsWithDecodingExceptions` option from pipeline options and pass it to the `UngroupedWindmillReader` constructor. * Updated the `UngroupedWindmillReaderIterator` constructor to pass the `skipDecodingErrors` flag to its superclass. * Changed `context.getSerializedKey().newInput()` to `checkNotNull(context.getSerializedKey()).newInput()` for null safety. * **runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillKeyedWorkItem.java** * Imported `Internal`, `Logger`, and `LoggerFactory`. * Added `@Internal` annotation to the class and removed old `@SuppressWarnings`. * Added `LOG` static field for logging. * Added `skipExceptions` field and an overloaded constructor to accept this flag. * Refactored element decoding logic into a new private method `parseElem` which handles `IOException` by logging and returning null if `skipExceptions` is true. * Modified `elementsIterable()` to use `parseElem` and filter out null results. * Added `@SuppressWarnings({"unchecked"})` to `FakeKeyedWorkItemCoder` constructor. * **runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillReaderIteratorBase.java** * Imported `Logger` and `LoggerFactory`. * Removed `@SuppressWarnings` annotation. * Added `skipDecodingErrors` field and `LOG` static field for logging. * Updated constructor to accept the `skipDecodingErrors` flag. * Modified the `advance()` method to catch `IOException` during message decoding; if `skipDecodingErrors` is true, the error is logged, and the element is skipped. * **runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindowingWindmillReader.java** * Imported `Objects`, `DataflowStreamingPipelineOptions`, and `ValueProvider`. * Added `skipDecodingExceptions` field and updated constructors to accept this new boolean flag. * Modified the `Factory` class to retrieve the `skipInputElementsWithDecodingExceptions` option from pipeline options and pass it to the `WindowingWindmillReader.create` method. * **runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java** * Added `testSkipInputElementsWithDecodingExceptions` test case to verify that corrupted input elements are skipped when the new option is enabled. * **runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillKeyedWorkItemTest.java** * Imported `Iterables` and `assertTrue`. * Added `testElementIterationWithSkipEnabled`, `testElementIterationSkips`, and `testElementIterationAllSkips` test cases to validate the skipping of corrupted elements in `WindmillKeyedWorkItem`. * Added `addCorruptedElement` helper method for tests. * **runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillReaderIteratorBaseTest.java** * Imported `ThreadLocalRandom` and `CoderException`. * Modified `TestWindmillReaderIterator.decodeMessage` to throw `CoderException` for negative timestamps to simulate decoding errors. * Added `testSkipErrors` to verify the behavior of skipping errors in the iterator. * Updated `testForMessageBundleCounts` to accept a `skipErrors` parameter and inject corrupted messages randomly. * **sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/PaneInfo.java** * Added validation in `Encoding.fromTag` to check for valid index range and throw `CoderException` for invalid pane encodings. </details> <details> <summary><b>Activity</b></summary> * The pull request introduces a new pipeline option to handle decoding exceptions in streaming Dataflow jobs. * The core logic for skipping corrupted input elements has been implemented across several reader and work item classes. * Extensive unit tests have been added to ensure the new functionality works as expected, covering scenarios with valid, corrupted, and entirely corrupted input streams. * The author has also addressed minor code quality improvements, such as updating `@SuppressWarnings` annotations and adding null checks. </details> <details> <summary><b>Using Gemini Code Assist</b></summary> <br> The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips. <b>Invoking Gemini</b> 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. <b>Customization</b> 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](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). <b>Limitations & Feedback</b> Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with :thumbsup: and :thumbsdown: 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](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG). <b>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](https://cloud.google.com/products/gemini/code-assist).</b> </details> [^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487). -- 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]
