[FLINK-11693] Add KafkaSerializationSchema that directly uses ProducerRecord

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Implemented
• Fix Version/s: 1.9.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available

Environment

Kafka 0.11.0.0
Flink 1.4.0
flink-connector-kafka-0.11_2.11

Release Note

The modern `FlinkKafkaProducer` supports a new `KafkaSerializationSchema` that will replace `KeyedSerializationSchema` in the long run. This new schema allows directly generating Kafka `ProducerRecords` for sending to Kafka, therefore enabling the user to use all available Kafka features (in the context of Kafka records).

• Assignee: Aljoscha Krettek
• Reporter: Mohammad Abareghi
• Created: 20/Feb/19 15:36
• Updated: 28/Jun/19 07:32
• Resolved: 28/Jun/19 07:32

Description

This allows using all fields that Kafka offers on ProducerRecord, for example, timestamps and headers.

Old Text

Kafka has introduced notion of Header for messages in version 0.11.0.0 https://issues.apache.org/jira/browse/KAFKA-4208.

But flink-connector-kafka-0.11_2.11 which supports kafka 0.11.0.0 ignores headers when consuming kafka messages.

It would be useful in some scenarios, such as distributed log tracing, to support message headers to FlinkKafkaConsumer011 and FlinkKafkaProducer011.

Issue Links

• is a clone of: FLINK-8354 Add KafkaDeserializationSchema that directly uses ConsumerRecord
• is duplicated by: FLINK-11303 Utilizing Kafka headers for serialization and deserialization
• is duplicated by: FLINK-11647 Use message headers in flink kafka connectors
• links to: GitHub Pull Request #6615
• links to: GitHub Pull Request #7782
• links to: GitHub Pull Request #8535