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 = maxTs.max(timestamp)
timestamp
}
override def getCurrentWatermark: WatermarkOld = {
println(s"event watermark: ${maxTs - maxTimeLag}")
new WatermarkOld(maxTs - maxTimeLag)
}
```
The underlying events come from a kafka source, and are then handed to a
process function. The implementation is irrelevant for the question, I will
just share the relevant bit:
```scala
override def processElement(
event: MyEvent,
ctx: KeyedProcessFunction[String, MyEvent, MyEvent]#Context,
out: Collector[StreamEvent]
): Unit = {
println(
s"In process function, got event: $event, ctx.timestamp:
${ctx.timestamp}, currentWatermark: ${ctx.timerService.currentWatermark}"
)
...
}
```
When I run this app on a real kubernetes cluster using a kafka source topic
having idle partitions, the watermark is held back to 0 as expected:
```
In process function, got event: xxx, ctx.timestamp: 1601475710619,
currentWatermark: 0
```
I can also see these logs generated in the watermarker:
```
event watermark: 1601475710619
event watermark: 0
event watermark: 1601475710619
event watermark: 0
```
The funny thing is that when I run the same application locally on IntelliJ,
and also have idle kafka partitions for the same topic, I am getting also
the above logs from the watermarker, with the watermark oscillating between
0 and the ts of the latest received element, since `maxLag = 0`. However,
quite unexpectedly for me, the logs from the process function show that the
watermark is yet advancing:
```
In process function, got event: xxx, ctx.timestamp: 1601475710619,
currentWatermark: 1601475710618
```
Why is this happening? FYI, I am using Flink 1.10 with the environment
parallelism set to 2 and event time semantics in both cases.
--
Sent from:
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
Locked
