Search code examples
javascalaapache-kafkaapache-flinkkafka-consumer-api

How to get Kafka timestamp for a record in Flink?

I have a Flink application that uses the FlinkKafkaConsumer.

I am interested in retrieving the Kafka timestamp for a given record/offset using the KeyedDeserializationSchema. It seems that the topic, partition, offset, and message are available.

How can the timestamp be obtained?

Solution

  • You can simply use ConsumerRecord.timestamp() - the timestamp in milliseconds since beginning of the epoch (midnight Jan 1, 1970 (UTC)). The timestamp type is indicated in timestampType().