azagrebin commented on a change in pull request #6615: [FLINK-8354]
[flink-connectors] Add ability to access and provider Kafka headers
URL: https://github.com/apache/flink/pull/6615#discussion_r257965533
##########
File path:
flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchema.java
##########
@@ -55,4 +57,13 @@
* @return null or the target topic
*/
String getTargetTopic(T element);
+
+ /**
+ *
+ * @param element The incoming element to be serialized
+ * @return collection of headers (maybe empty)
+ */
+ default Iterable<Map.Entry<String, byte[]>> headers(T element) {
Review comment:
The change for `KafkaDeserialisationSchema` might be simple if we do it like
this:
```
// new API, that we could now use everywhere in Kafka code instead of
KeyedDeserializationSchema
interface KafkaDeserialisationSchema<T> {
T deserialize(ConsumerRecord<byte[], byte[]> record);
}
// old API, stays almost the same as we already have it in the PR and also
supports the new API
@deprecated
interface KeyedDeserializationSchema<T> extends
KafkaDeserialisationSchema<T> {
T deserialize(byte[] messageKey, byte[] message, String topic, int
partition, long offset);
default T deserialize(ConsumerRecord<byte[], byte[]> record) {
return deserialize(record.key(), record.value(), record.topic(),
record.partition(), record.offset());
}
}
```
Do you think it is quick enough change?
Ok, I see your point with tests regarding the splitting. The thing is that
we are interested in merging this PR this week before feature freeze for 1.8
release (planned this Friday). Let's see then how it goes and how much we can
manage. Do you have time this week to work on this?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
