You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Alex Gout (Jira)" <ji...@apache.org> on 2023/02/14 14:12:00 UTC
[jira] [Updated] (FLINK-31049) Add support for Kafka record headers to KafkaSink
[ https://issues.apache.org/jira/browse/FLINK-31049?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Alex Gout updated FLINK-31049:
------------------------------
Description:
The default org.apache.flink.connector.kafka.sink.KafkaSink does not support adding Kafka record headers. In some implementations, downstream consumers might rely on Kafka record headers being set.
A way to add KafkaHeaders would be to create a custom
KafkaRecordSerializationSchemaBuilder
The code responsible for creating the Kafka record involves org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaWrapper where the RecordProducer is created.
It is relatively simple to add support for record headers by adding a "HeaderProducer" next to the key and value serializers and using the appropriate RecordProducer constructor.
The issue was discussed [here|https://lists.apache.org/thread/shlbbcqho0q9w5shjwdlscnsywjvbfro].
was:
The default org.apache.flink.connector.kafka.sink.KafkaSink does not support adding Kafka record headers. In some implementations, downstream consumers might rely on Kafka record headers being set.
The code responsible for creating the Kafka record involves org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaWrapper where the RecordProducer is created.
It is relatively simple to add support for record headers by adding a "HeaderProducer" next to the key and value serializers and using the appropriate RecordProducer constructor.
The issue was discussed [here|https://lists.apache.org/thread/shlbbcqho0q9w5shjwdlscnsywjvbfro].
> Add support for Kafka record headers to KafkaSink
> -------------------------------------------------
>
> Key: FLINK-31049
> URL: https://issues.apache.org/jira/browse/FLINK-31049
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Kafka
> Reporter: Alex Gout
> Priority: Minor
> Labels: KafkaSink
> Original Estimate: 48h
> Remaining Estimate: 48h
>
> The default org.apache.flink.connector.kafka.sink.KafkaSink does not support adding Kafka record headers. In some implementations, downstream consumers might rely on Kafka record headers being set.
>
> A way to add KafkaHeaders would be to create a custom
> KafkaRecordSerializationSchemaBuilder
> The code responsible for creating the Kafka record involves org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaWrapper where the RecordProducer is created.
> It is relatively simple to add support for record headers by adding a "HeaderProducer" next to the key and value serializers and using the appropriate RecordProducer constructor.
>
> The issue was discussed [here|https://lists.apache.org/thread/shlbbcqho0q9w5shjwdlscnsywjvbfro].
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)