[jira] [Created] (FLINK-11303) Utilizing Kafka headers for serialization and deserialization

["Allen Wang (JIRA)" <ji...@apache.org>]

Allen Wang created FLINK-11303:
----------------------------------

             Summary: Utilizing Kafka headers for serialization and deserialization
                 Key: FLINK-11303
                 URL: https://issues.apache.org/jira/browse/FLINK-11303
             Project: Flink
          Issue Type: Improvement
          Components: Kafka Connector
            Reporter: Allen Wang


Kafka introduces headers in producer and consumer record since version 0.11. This is the high level description: [https://cwiki.apache.org/confluence/display/KAFKA/KIP-82+-+Add+Record+Headers]

However, current Flink Kafka connector simply ignores the headers. This will make it hard to integrate with the Kafka ecosystem where other Kafka clients make use of the headers.

 

I propose to support headers in Flink by modifying the following API:
 * In KeyedSerializationSchema, add
{code:java}
List<Tuple2<String, byte[]>> getHeaders(T element)
{code}

 * In KeyedDeserializationSchema, add
{code:java}
T deserailize(byte[] messageKey, byte[] message, List<Tuple2<String, byte[]>> headers, String topic, int partition, long offset) throws IOException{code}

 

These new methods will be invoked by FlinkKafkaProducer and KafkaFetcher in the serialization and deserialization process. If backward compatibility is a concern, we can add default implementation to these methods where headers are ignored.

 

If backward compatiblity 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)