Repository navigation
Conversation
Summary of ChangesHello, 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 enhances the Streaming Data Generator's Kafka sink by enabling support for record keys and headers. By introducing an optional envelope structure in the schema, users can now specify keys and headers alongside the message payload. The implementation migrates the sink to use KafkaIO.writeRecords(), providing better integration with Kafka's producer record features while maintaining backward compatibility for existing keyless message generation. 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 the 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 counterproductive. 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. Footnotes
|
![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 Kafka as a sink in the streaming data generator, allowing messages to be written with optional keys and headers using an envelope schema format. It updates the Kafka writing transform to process these envelopes and adds comprehensive unit tests. The review feedback suggests renaming the production base class FakeKafkaMessageFn to a more appropriate name like BaseKafkaMessageFn to avoid confusion, and marking non-serializable Avro fields in AvroKafkaMessageFn as transient to prevent serialization errors during pipeline execution.
The Kafka sink wrote every message keyless and without headers because it
used KafkaIO.write().values(). The sink now produces
ProducerRecord<String, byte[]> through KafkaIO.writeRecords() and lets the
schema opt in to a key and headers with an envelope, mirroring the Pub/Sub
payload/attributes convention:
{"payload": {...}, "key": "{{get('accountId')}}", "headers": {...}}
- payload is written as the record value (JSON, or Avro binary for
outputType=AVRO; the Avro output was previously Arrays.toString(bytes)).
- key becomes the record key (StringSerializer); absent -> keyless.
- each headers member becomes a record header (UTF-8 text / compact JSON).
- messages without an envelope are written exactly as before.
KafkaIO.Write is still configured the same way (including
withGCPApplicationDefaultCredentials() for Managed Kafka) and its
WriteRecords transform is applied to the records. Adds unit tests and
README documentation.
08f48a0 to
733e38e
Compare
Codecov Report❌ Patch coverage is
❌ Your patch check has failed because the patch coverage (68.33%) is below the target coverage (80.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## main #4482 +/- ##
=============================================
+ Coverage 38.03% 57.04% +19.01%
- Complexity 518 7619 +7101
=============================================
Files 195 1161 +966
Lines 11885 73930 +62045
Branches 1213 8709 +7496
=============================================
+ Hits 4520 42171 +37651
- Misses 6994 28951 +21957
- Partials 371 2808 +2437
🚀 New features to boost your workflow:
|
The Kafka sink wrote every message keyless and without headers because it used
KafkaIO.write().values(). The sink now producesProducerRecord<String, byte[]>throughKafkaIO.writeRecords()and lets the schema opt in to a key and headers with an envelope, mirroring the Pub/Sub payload/attributes convention:{"payload": {...}, "key": "{{get('accountId')}}", "headers": {...}}Arrays.toString(bytes)).StringSerializer); absent -> keyless.KafkaIO.Writeis still configured the same way (including withGCPApplicationDefaultCredentials() for Managed Kafka) and its WriteRecords transform is applied to the records.