2

I am reading the book Stream Processing with Apache Flink and it is stated that “As of version 0.10.0, Kafka supports message timestamps. When reading from Kafka version 0.10 or later, the consumer will automatically extract the message timestamp as an event-time timestamp if the application runs in event-time mode*” So inside a processElement function the call context.timestamp() will by default return the kafka message timestamp? Coul you please provide a simple example on how to implement AssignerWithPeriodicWatermarks/AssignerWithPunctuatedWatermarks that extract (and builds watermarks) based on the consumed kafka message timestamp.

If I am using TimeCharacteristic.ProcessingTime, would ctx.timestamp() return the processing time and in such case would it be similar to context.timerService().currentProcessingTime() .

Thank you.

David Anderson
  • 39,434
  • 4
  • 33
  • 60
Mazen Ezzeddine
  • 662
  • 1
  • 8
  • 28

1 Answers1

4

The Flink Kafka consumer takes care of this for you, and puts the timestamp where it needs to be. In Flink 1.11 you can simply rely on this, though you still need to take care of providing a WatermarkStrategy that specifies the out-of-orderness (or asserts that the timestamps are in order):

FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>(...);
myConsumer.assignTimestampsAndWatermarks(
    WatermarkStrategy.
        .forBoundedOutOfOrderness(Duration.ofSeconds(20)));

In earlier versions of Flink you had to provide an implementation of a timestamp assigner, which would look like this:

public long extractTimestamp(Long element, long previousElementTimestamp) {
    return previousElementTimestamp;
}

This version of the extractTimestamp method is passed the current value of the timestamp present in the StreamRecord as previousElementTimestamp, which in this case will be the timestamp put there by the Flink Kafka consumer.

Flink 1.11 docs
Flink 1.10 docs

As for what is returned by ctx.timestamp() when using TimeCharacteristic.ProcessingTime, this method returns NULL in that case. (Semantically, yes, it is as though the timestamp is the current processing time, but that's not how it's implemented.)

David Anderson
  • 39,434
  • 4
  • 33
  • 60
  • is there a way I can specify the record's timestamp rather than Kakfa record timestamp? For instance, I want the Kafka consumer to look at my shopping website order timestamp from the client to advance the watermark. Below is the code I've tried – ksp0422 Nov 03 '20 at 12:56
  • ```WatermarkStrategy.forBoundedOutOfOrderness[Order](Duration.ofMillis(100)).withTimestampAssigner(new SerializableTimestampAssigner[Order] { override def extractTimestamp(order: Order, recordTimestamp: Long): Long = { order.getTimestamp }})}``` – ksp0422 Nov 03 '20 at 12:58
  • @DavideAnderson Since *When reading from Kafka version 0.10 or later, the consumer will automatically extract the message timestamp as an event-time timestamp if the application runs in event-time mode*, so all downstream Flink operators in event-time mode will see all the event-times as strictly increasing, because Kafka ConsumerRecord's timestamp seems its ingestion-time. So seems to me there won't be any out-of-orderness after reading events from a Kafka topic per partition, therefore only the other MonotoneTime watermark strategy make sense in these cases... Any comment? Thanks in advance! – mohottnad Apr 01 '22 at 03:41
  • 1
    While Kafka consumer records often have log-append timestamps set by the brokers, those timestamps could've been set by the upstream producers, in which case they might not be strictly ascending. – David Anderson Apr 02 '22 at 08:05