zhiliang-wu opened a new pull request, #11112:
URL: https://github.com/apache/seatunnel/pull/11112
<!--
Thank you for contributing to SeaTunnel! Please make sure that your code
changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
## Contribution Checklist
- Make sure that the pull request corresponds to a [GITHUB
issue](https://github.com/apache/seatunnel/issues).
- Name the pull request in the form "[Feature] [component] Title of the
pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
- Minor fixes should be named following this pattern: `[hotfix] [docs] Fix
typo in README.md doc`.
-->
### Purpose of this pull request
<!-- Describe the purpose of this pull request. For example: This pull
request adds checkstyle plugin.-->
This PR introduces a new `kafka_message_value_fields` configuration option
to the Kafka Sink connector.
Currently, all fields from an upstream row (excluding those in
`kafka_headers_fields`) are serialized into the Kafka message value. This
causes issues in pipelines where a specific column (like `message_key`) is used
for partitioning or routing, but the user does not want that routing key
serialized into the final message body payload.
By configuring `kafka_message_value_fields`, users can explicitly dictate
exactly which fields get written into the Kafka message value. This cleanly
decouples partition keys from the output payload, allowing users to route by
`message_key` but only output `message_value` into the Kafka body.
### Does this PR introduce _any_ user-facing change?
Yes. It adds a new optional parameter `kafka_message_value_fields` to the
Kafka Sink.
**Previous behavior:**
All upstream fields (except explicitly designated headers) were serialized
into the message value.
**New behavior:**
- If `kafka_message_value_fields` is specified, *only* the fields listed in
this array are serialized into the message value.
- If it is not specified, the connector falls back to the previous default
behavior.
- A strict validation check has been added: if a user configures the exact
same field in both `kafka_message_value_fields` and `kafka_headers_fields`, the
job will immediately fail during initialization with an
`IllegalArgumentException` to prevent accidental payload duplication.
- English and Chinese (`zh`) documentation for the Kafka Sink has been
updated to explain the new parameter.
### How was this patch tested?
1. **Unit Tests Added:** Added `testMessageValueFields()` in
`DefaultSeaTunnelRowSerializerTest` to strictly verify that only the requested
fields are serialized into the `ProducerRecord` value payload and unrequested
fields (like `message_key`) are properly omitted.
2. **Backward Compatibility Tests:** Verified that existing behaviors
regarding `partition_key_fields` and `kafka_headers_fields` remain completely
unaffected when `kafka_message_value_fields` is null.
3. **Local E2E Verification:** Manually tested via local SeaTunnel jobs
using a `MaxCompute -> Kafka` pipeline to ensure the payload properly
restricted to the specified fields.
4. **Build Passing:** Code has been formatted using `./mvnw spotless:apply`
and tests successfully pass using `./mvnw clean test -pl
seatunnel-connectors-v2/connector-kafka`.
### Check list
* [ ] If any new Jar binary package adding in your PR, please add License
Notice according
[New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
* [x] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs
* [ ] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR.
* [ ] If you are contributing the connector code, please check that the
following files are updated:
1. Update
[plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties)
and add new connector information in it
2. Update the pom file of
[seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml)
3. Add ci label in
[label-scope-conf](https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml)
4. Add e2e testcase in
[seatunnel-e2e](https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/)
5. Update connector
[plugin_config](https://github.com/apache/seatunnel/blob/dev/config/plugin_config)
--
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]
