

Re: TimerService/Watermarks and Checkpoints

Thanks for the explanation. I looked at this metric closely and noticed there are some events arriving in out of order. The hypothesis I have is, when the job is restarted, all of the small out of order chunks add up and show a significant number. The graph below shows the number of out of order events every min. The job was started with new state at 11:53 am and then restarted with the previous checkpoint at 1:24 pm. 

That said, after restart the out of order events number is very high though :thinking_face:

On Wed, May 30, 2018 at 1:55 PM, Fabian Hueske <fhueske@xxxxxxxxx> wrote:
Hi Nara and Sihua,

That's indeed an unexpected behavior and it would be good to identify the reason for the late data.

As Sihua said, watermarks are currently not checkpointed and reset to Long.MIN_VALUE upon restart.
AFAIK, the main reason why WMs are not checkpointed is that the special type of operator state that is required for this (union-list state) wasn't available when the mechanism was implemented.
I think there are plans to address this shortcoming (see FLINK-5601 [1]).

Best, Fabian

2018-05-30 19:00 GMT+02:00 Narayanan Arunachalam <narayanan.arunachalam@gmail.com>:
Thanks Sihua. If it's reset to Long.MIN_VALUE I can't explain why outOfOrderEvents are reported. Because the event time on the data will always be greater than Long.MIN_VALUE.

Following are the steps to reproduce this scenario.
- A source to produce events with timestamps that is increasing for every event produced
- Use TimeCharacteristic.EventTime
- Use BoundedOutOfOrdernessTimestampExtractor with maxOutOfOrderness set to 60s. 
- Enable checkpoints
- ProcessFunction impl to report a counter to some metrics backend when the timestamp of the event is less than currentWatermark
- No out of order events will be reported initially. After few checkpoints are created, cancel and restart the job from a previous checkpoint.

Note: The event stream really doesn't have out of order data. Job restart from a checkpoint causes this artificial out of order events because of the watermark value.


On Tue, May 29, 2018 at 7:54 PM, sihua zhou <summerleafs@xxxxxxx> wrote:
Hi Nara,

yes, the watermark in TimerService is not covered by the checkpoint, everytime the job is restarted from a previous checkpoint, it is reset to Long.MIN_VALUE. I can see it a bit tricky to cover it into the checkpoint, especially when we need to support rescaling(it seems not like a purely keyed or a operate state), maybe @Stefan or @Aljoscha could give you more useful information about why it wasn't covered by the checkpoint.

Best, Sihua


Is it possible the watermark in TimerService not getting reset when a job is restarted from a previous checkpoint? I would expect the watermark in a TimerService also to go back in time.

I have the following ProcessFunction implementation.

  override def processElement(
    e: TraceEvent,
    ctx: ProcessFunction[
    out: Collector[Trace]
  ): Unit = {

    if (e.getAnnotationTime() < ctx.timerService().currentWatermark()) {
    } else {

I am noticing the outOfOrderEvents are reported until new events are read in to the stream since the last restart.