Awesome David, thanks for clarifying!
--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
If you were to use per-partition watermarking, which you can do by
calling assignTimestampsAndWatermarks directly on the Flink Kafka consumer
[1], then I believe the idle partition(s) would consistently hold back the
overall watermark.
With per-partition watermarking, each Kafka source task will
I am considering this watermarker:
```scala
class MyWatermarker(val maxTimeLag: Long = 0)
extends AssignerWithPeriodicWatermarks[MyEvent] {
var maxTs: Long = 0
override def extractTimestamp(e: MyEvent, previousElementTimestamp: Long):
Long = {
val timestamp = e.timestamp
maxTs =
