Hi everyone, how do I retrieve timestamps from records received from confluent Kafka. I want to assign these timestamps using

.withTimestampAssigner()

on my Confluent Kafka sources. Bonus question, how do I know if flink already extracted timestamps from the records. The documentation, says that if timestamps are set in the headers of my records, Flink automatically retrieves them, does anyone know how I can check this?

Your timestamp assigner will look like this:

.withTimestampAssigner(
  (event, streamRecordTimestamp) -> ...)

The streamRecordTimestamp passed in, if any, will be the timestamp extracted from the Kakfa headers.

So if it exists in the headers I don’t need to return anything inside the function

If you provide a timestamp assigner it should return the timestamp you want to use.

I apologize if this is repetitive but if I don’t want to supply a timestamp assigner and there are timestamps in my records, do I still need to return anything

What do you want to do if the headers don’t have timestamps?

I don’t have time stamps in the payload but when I look at the messages on the Confluent topics they look to have timestamps in the metadata

Then you don’t need a timestamp assigner at all. However, if you want to test that the timestamps in the headers are being used, you can do that with a timestamp assigner, but it should return the timestamp.

Thanks @David Anderson, with your suggestions my watermarks seem to be advancing. Now I have to find out why my stream join is still not producing any output.