Hi Dawid, I'm working with Max on the project
Our code for the TimestampAndWatermarkAssigner is:
```
class TimestampAndWatermarkAssigner(val maxLateness: Long) extends
AssignerWithPeriodicWatermarks[Row] {
override def extractTimestamp(element: Row, previousElementTimestamp:
Long): Long = {
element.minTime
}
override def getCurrentWatermark(): Watermark = {
new Watermark(System.currentTimeMillis() - maxLateness)
}
}
```
Where Row is a class representing the incoming JSON object coming from
Kafka, which includes the timestamp
Thanks,
-Ethan
--
View this message in context:
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Issues-with-Event-Time-and-Kafka-tp12061p12090.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at
Nabble.com.
