[FLINK-8354] Add KafkaDeserializationSchema that directly uses ConsumerRecord

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Implemented
• Fix Version/s: 1.8.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: For the Flink KafkaConsumers, we introduced a new KafkaDeserializationSchema that gives direct access to the Kafka ConsumerRecord. This subsumes the KeyedSerializationSchema functionality, which is deprecated but still available for now.
• Assignee: Aegeaner
• Reporter: Mohammad Abareghi
• Created: 03/Jan/18 13:14
• Updated: 31/Jul/19 09:58
• Resolved: 22/Feb/19 11:23

Description

This allows using all fields that Kafka offers on ConsumerRecord, 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 cloned by: FLINK-11693 Add KafkaSerializationSchema that directly uses ProducerRecord
• is duplicated by: FLINK-8500 Get the timestamp of the Kafka message from kafka consumer
• is duplicated by: FLINK-11303 Utilizing Kafka headers for serialization and deserialization
• is duplicated by: FLINK-11647 Use message headers in flink kafka connectors
• relates to: FLINK-13507 kafka consumer not able to distinguish topic name after add source
• links to: GitHub Pull Request #6615
• links to: GitHub Pull Request #7781