Re: flink operator's latency metrics continues to increase.

[Till Rohrmann <tr...@apache.org>]

Hi Suxing Lee,

thanks for reaching out to me. I forward this mail also to the user mailing
list because it could be interesting for others as well.

Your observation could indeed be an indicator for a problem with the
latency metric. I quickly checked the code and on the first glance it looks
right to me that we increase the nextTimestamp field by period in
RepeatedTriggerTask because we schedule this task at a fixed rate in
SystemProcessingTimeService#scheduleAtFixedRate. Internally this method
calls ScheduledThreadPoolExecutor#scheduleAtFixedRate which uses
System.nanoTime to schedule tasks repeatedly. In fact, the same logic will
be used by the ScheduledThreadPoolExecutor#ScheduledFutureTask. If a GC
pause or another stop the world event happens, this should only affect one
latency metric and not all (given that System.nanoTime continues to
increase) because the next will be scheduled faster since System.nanoTime
might have progressed more.

What could be a problem is that we compute the latency by
System.currentTimeMillis - marker.getMarkedTime. I think there is no
guarantee that System.currentTimeMillis and System.nanoTime don't drift
apart. Especially if they are executed on different machines. This is
something which we could check.

This link [1] explains the drift problem a bit more in detail.

In any case, I would suggest to open a JIRA issue to report this problem.

[1] https://bugs.java.com/bugdatabase/view_bug.do?bug_id=6519418

Cheers,
Till

On Mon, Dec 17, 2018 at 2:37 PM Suxing Lee <91...@qq.com> wrote:

> Hi Till Rohrmann,
>
> I was running flink 1.5.5,  and I use prometheus to collect metrics to
> check latency of my jobs.
> But sometimes I observerd that  the operator's latency metrics continues
> to increase in my job.
> The operator's latency time is increased by approximately 2.7 minutes per
> day (please see the attached screenshots)
>
> my job's logic is simple，just distribute data from kafkaSource to
> bucketingSink.
> so  I check the consumer offsets in kafka for  consumer group, I also
> check the latest data in hdfs . in fact, there is no serious latency in my
> job.
>
> I notice that the statistical method of latency is currentTimeMillis minus
> LatencyMarker's markedTime.
> but LatencyMarker's timestamp come from RepeatedTriggerTask's
> nextTimestamp which compute timestamp by plus a period（default  value is 2s
> before v1.5.5），the nextTimestamp will be delay when JVM GC or linux
> preemptive scheduling happened. as time increases，the nextTimestamp is much
> later than the current time ( I had verify this result via  the JVM Heap
> Dump).
>
> we can avoid the above situation by directly using linux's NTP to
> guarantee accuracy，not need to compute timestamp by process.
> I'm not very familiar with  SystemProcessingTimeService. Is there some
> detail I have not think about?
>
>
> Best regards and thanks for your help.
> Suxing Lee
>
>
>
>
>