Add support for UDF function transformations in PubSubToRedis template#3338
Add support for UDF function transformations in PubSubToRedis template#3338wladabutenko wants to merge 3 commits intoGoogleCloudPlatform:mainfrom
Conversation
Summary of ChangesHello @wladabutenko, 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 significantly enhances the PubSubToRedis Dataflow template by introducing optional User-Defined Function (UDF) capabilities. Users can now apply custom JavaScript transformations to their Pub/Sub messages before they are stored in Redis, providing greater flexibility in data processing. Additionally, robust error handling has been integrated through a dead-letter topic, ensuring that any messages failing UDF processing or Redis writes are captured for further analysis. The changes also include updated parameter validation, comprehensive documentation, and new integration tests to ensure the reliability and correctness of the new features. Highlights
Changelog
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
|
|
/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 adds support for user-defined JavaScript functions (UDFs) and a dead-letter topic to the Pub/Sub to Redis template. The changes are well-structured and include updates to the pipeline code, documentation, and tests. My review focuses on improving documentation accuracy, ensuring code correctness, and enhancing test coverage. I've identified a few areas for improvement:
- The descriptions for the new
deadletterTopicparameter in the README, Javadoc, and Terraform configuration are slightly misleading and should be updated to reflect that only UDF failures are handled. - There's a potential bug in character encoding when creating a
PubsubMessagefrom the transformed payload. - The new integration tests don't cover the UDF and dead-letter functionality, and one of the new unit tests is incomplete.
Overall, this is a great addition to the template. Addressing these points will make it more robust and easier to use.
| FailsafeElement<PubsubMessage, String> element = c.element(); | ||
| c.output( | ||
| new PubsubMessage( | ||
| element.getPayload().getBytes(), |
There was a problem hiding this comment.
Calling getBytes() on a String without specifying a charset can lead to unexpected behavior as it uses the platform's default charset. Since the payload is consistently handled as UTF-8 in other parts of the code (e.g., in PubsubMessageToFailsafeElementFn), it's best to explicitly use StandardCharsets.UTF_8 here for consistency and correctness.
| element.getPayload().getBytes(), | |
| element.getPayload().getBytes(java.nio.charset.StandardCharsets.UTF_8), |
There was a problem hiding this comment.
Seems like a reasonable concern.
I think we should have helpers for these PubsubMessage<->Failsafe manipulations in a shareable location:
I am seeing these already:
https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/common/src/main/java/com/google/cloud/teleport/v2/transforms/PubSubToFailSafeElement.java
https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/PubsubMessageToFailsafeElementFn.java
Then, helpers should use consistent encodings.
| * **redisHost**: The Redis database host. For example, `your.cloud.db.redislabs.com`. Defaults to: 127.0.0.1. | ||
| * **redisPort**: The Redis database port. For example, `12345`. Defaults to: 6379. | ||
| * **redisPassword**: The Redis database password. Defaults to `empty`. | ||
| * **deadletterTopic**: The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation or cannot be written to Redis are forwarded here. For example, `projects/<PROJECT_ID>/topics/<TOPIC_NAME>`. |
There was a problem hiding this comment.
The description for deadletterTopic states that messages that "cannot be written to Redis" are forwarded to this topic. Based on the current implementation, only messages that fail the UDF transformation are sent to the dead-letter topic. Failures during the write to Redis are not handled this way. To avoid confusion, please update the description to accurately reflect that only UDF failures are sent to the dead-letter topic.
| * **deadletterTopic**: The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation or cannot be written to Redis are forwarded here. For example, `projects/<PROJECT_ID>/topics/<TOPIC_NAME>`. | |
| * **deadletterTopic**: The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation are forwarded here. For example, `projects/<PROJECT_ID>/topics/<TOPIC_NAME>`. |
| helpText = | ||
| "The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation or cannot be written to Redis are forwarded here.", |
There was a problem hiding this comment.
The helpText for the deadletterTopic parameter states that messages that "cannot be written to Redis" are forwarded to this topic. However, the implementation only sends messages that fail the UDF transformation to the dead-letter topic. Please update the documentation to be consistent with the implementation.
| helpText = | |
| "The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation or cannot be written to Redis are forwarded here.", | |
| helpText = | |
| "The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation are forwarded here.", |
| // SkipDirectRunnerTest: PubsubIO doesn't trigger panes on the DirectRunner. | ||
| @Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) | ||
| @RunWith(JUnit4.class) | ||
| public final class PubSubToRedisIT extends TemplateTestBase { |
There was a problem hiding this comment.
These integration tests are a good start for the Redis sinks. However, they don't cover the new UDF transformation and dead-lettering functionality, which is the main feature of this pull request. To ensure the end-to-end flow is working as expected, please consider adding integration tests for the following scenarios:
- A successful UDF transformation.
- A failing UDF transformation, verifying that the message is sent to the dead-letter topic.
- The case where no UDF is specified (passthrough).
This will significantly improve the test coverage for the new features.
| // Verify that the data was written to Redis Streams (additional verification logic may be | ||
| // needed) |
There was a problem hiding this comment.
This test case for the Redis Streams sink appears to be incomplete. It runs the pipeline but does not include any assertions to verify that the data was actually written to the Redis stream. Please add assertions to confirm the expected outcome, for example by checking the stream length with client.xlen() or inspecting its content with client.xrange().
|
|
||
| variable "deadletterTopic" { | ||
| type = string | ||
| description = "The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation or cannot be written to Redis are forwarded here. For example, `projects/<PROJECT_ID>/topics/<TOPIC_NAME>`" |
There was a problem hiding this comment.
The description for the deadletterTopic variable states that messages that "cannot be written to Redis" are forwarded to this topic. This is inconsistent with the implementation, which only handles UDF failures. Please update the description to accurately reflect the functionality.
description = "The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation are forwarded here. For example, `projects/<PROJECT_ID>/topics/<TOPIC_NAME>`"
There was a problem hiding this comment.
thoughts on this one? what do other templates do?
There was a problem hiding this comment.
I think this is due to the nature of Redis as it treats the data as an binary-safe, opaque block of text. So for STRING_SINK type it does not validate whether the string malformed/correct, it's takes the raw characters and stores them as a Redis string value.
Based on v2/pubsub-to-redis/src/main/java/com/google/cloud/teleport/v2/templates/transforms/MessageTransformation.java the entire message payload mapped into fieldKey
There was a problem hiding this comment.
is there a failure mode where a message is not written to Reddis for whatever reason? if so, will such message be forwarded to DLQ?
There was a problem hiding this comment.
The comment (which I agree with) is that if only messages for which UDF failed to run are forwarded to DLQ, then the docstring should not say ...Messages that fail UDF transformation or cannot be written to Redis are forwarded here...
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #3338 +/- ##
============================================
+ Coverage 51.04% 51.08% +0.03%
- Complexity 5213 5598 +385
============================================
Files 994 994
Lines 60672 60725 +53
Branches 6651 6653 +2
============================================
+ Hits 30972 31022 +50
Misses 27527 27527
- Partials 2173 2176 +3
🚀 New features to boost your workflow:
|
wladabutenko
force-pushed
the
copy_pr_3262
branch
from
130bd43 to
27c0588
Compare
2 participants
Adding optional UDF support to the PubSubToRedis pipeline, enabling user-defined transformations before writing to Redis