Suspicious watermark of operators after restore from checkpoint

[Jan Lukavský]

Hi all,

I have just come across a weird state of operators after restore from
checkpoint. After the restore, two operators that are connected (i.e.
operator A is input of operator B) ended up with watermark of operator A
being less than watermark of operator B. I don't know how to explain
this. Can it be normal or does it signal a bug somewhere? If I
understand Flink's checkpointing correctly, the checkpoint barrier flows
from one operator to another, so the watermark should be aligned.

I'm running a Beam pipeline on Flink 1.8.1.

Am I missing something?

Many thanks for comments,

  Jan

[Kostas Kloudas-4]

Hi Jan,

Two pointers that may help you explain the behavior are the following:

1) If you have a custom watermark generator, I do not think that Flink
checks if it emits only monotonically
increasing watermarks. This is the responsibility of the generator itself.
This means that although you operator A
is topologically before operator B, operator A may have a smaller watermark
if your watermark generator allows so.

2) Flink currently does not checkpoint the last seen watermark (
https://issues.apache.org/jira/browse/FLINK-5601).
This means that after restoring, your (event) time is assumed to be
Long.Min until the first new watermark comes.
So if you observed late data not being late anymore or sth similar, then it
may not be that the two operators have
different watermarks but that after restoring event time rolls back to the
"beginning of time".

I hope this helps,
Kostas

On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]> wrote:

> Hi all,
>
> I have just come across a weird state of operators after restore from
> checkpoint. After the restore, two operators that are connected (i.e.
> operator A is input of operator B) ended up with watermark of operator A
> being less than watermark of operator B. I don't know how to explain
> this. Can it be normal or does it signal a bug somewhere? If I
> understand Flink's checkpointing correctly, the checkpoint barrier flows
> from one operator to another, so the watermark should be aligned.
>
> I'm running a Beam pipeline on Flink 1.8.1.
>
> Am I missing something?
>
> Many thanks for comments,
>
>   Jan
>
>

[Jan Lukavský]

Hi Kostas,

thanks for reaction, comments inline.

On 8/7/19 1:59 PM, Kostas Kloudas wrote:

> Hi Jan,
>
> Two pointers that may help you explain the behavior are the following:
>
> 1) If you have a custom watermark generator, I do not think that Flink
> checks if it emits only monotonically
> increasing watermarks. This is the responsibility of the generator itself.
> This means that although you operator A
> is topologically before operator B, operator A may have a smaller watermark
> if your watermark generator allows so.

I do generate watermarks by custom source, but I believe that the
generated sequence is monotonic. But still, I'm not sure, that even if
it was the case, that the generated watermark actually decreases, would
that mean, that downstream operator after source (operator A) would
actually "go back in time"?
>
> 2) Flink currently does not checkpoint the last seen watermark (
> https://issues.apache.org/jira/browse/FLINK-5601).
> This means that after restoring, your (event) time is assumed to be
> Long.Min until the first new watermark comes.
> So if you observed late data not being late anymore or sth similar, then it
> may not be that the two operators have
> different watermarks but that after restoring event time rolls back to the
> "beginning of time".

I actually didn't observe any wrong or unexpected behavior, exceptions
or wrong outputs. I just noticed this on Flink's WebUI and it looked
strange to me. Could it be just that the WebUI showed older watermark
for operator A? Strange was, that the watermarks were my screen long
enough to take a screenshot (so at least say 10 seconds displaying
watermark of operator A less than the one of operator B). Even if
watermarks are not checkpointed, would it still be possible for
watermark of operator B to be actually greater? I'm still confused of
how this could happen, because (in my understanding) output watermark of
operator A should be greater or equal to input watermark of B (because
it takes minimum of inputs).

Sorry if I'm too digging into this, but I don't like things I cannot
explain, as they might point out to some bugs somewhere. :-) Or that my
mental model it not aligned with reality.

Jan

>
> I hope this helps,
> Kostas
>
> On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]> wrote:
>
>> Hi all,
>>
>> I have just come across a weird state of operators after restore from
>> checkpoint. After the restore, two operators that are connected (i.e.
>> operator A is input of operator B) ended up with watermark of operator A
>> being less than watermark of operator B. I don't know how to explain
>> this. Can it be normal or does it signal a bug somewhere? If I
>> understand Flink's checkpointing correctly, the checkpoint barrier flows
>> from one operator to another, so the watermark should be aligned.
>>
>> I'm running a Beam pipeline on Flink 1.8.1.
>>
>> Am I missing something?
>>
>> Many thanks for comments,
>>
>>    Jan
>>
>>

[Kostas Kloudas-4]

Hi Jan,

After looking at the code, my point 1) is false for *intermediate* tasks
and if you are
using a watermark assigner. This means that in these cases, Flink checks
that the
"next" watermark is greater than the "previous" one.

But if your operator A is a source and you emit watermarks from the source,
then
it can happen that your watermark appears to go backwards on operator A,
but
operator B does the "correction" by discarding smaller watermarks. That can
explain
your observation.

Cheers,
Kostas

On Wed, Aug 7, 2019 at 2:30 PM Jan Lukavský <[hidden email]> wrote:

> Hi Kostas,
>
> thanks for reaction, comments inline.
>
> On 8/7/19 1:59 PM, Kostas Kloudas wrote:
> > Hi Jan,
> >
> > Two pointers that may help you explain the behavior are the following:
> >
> > 1) If you have a custom watermark generator, I do not think that Flink
> > checks if it emits only monotonically
> > increasing watermarks. This is the responsibility of the generator
> itself.
> > This means that although you operator A
> > is topologically before operator B, operator A may have a smaller
> watermark
> > if your watermark generator allows so.
> I do generate watermarks by custom source, but I believe that the
> generated sequence is monotonic. But still, I'm not sure, that even if
> it was the case, that the generated watermark actually decreases, would
> that mean, that downstream operator after source (operator A) would
> actually "go back in time"?
> >
> > 2) Flink currently does not checkpoint the last seen watermark (
> > https://issues.apache.org/jira/browse/FLINK-5601).
> > This means that after restoring, your (event) time is assumed to be
> > Long.Min until the first new watermark comes.
> > So if you observed late data not being late anymore or sth similar, then
> it
> > may not be that the two operators have
> > different watermarks but that after restoring event time rolls back to
> the
> > "beginning of time".
>
> I actually didn't observe any wrong or unexpected behavior, exceptions
> or wrong outputs. I just noticed this on Flink's WebUI and it looked
> strange to me. Could it be just that the WebUI showed older watermark
> for operator A? Strange was, that the watermarks were my screen long
> enough to take a screenshot (so at least say 10 seconds displaying
> watermark of operator A less than the one of operator B). Even if
> watermarks are not checkpointed, would it still be possible for
> watermark of operator B to be actually greater? I'm still confused of
> how this could happen, because (in my understanding) output watermark of
> operator A should be greater or equal to input watermark of B (because
> it takes minimum of inputs).
>
> Sorry if I'm too digging into this, but I don't like things I cannot
> explain, as they might point out to some bugs somewhere. :-) Or that my
> mental model it not aligned with reality.
>
> Jan
>
> >
> > I hope this helps,
> > Kostas
> >
> > On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]> wrote:
> >
> >> Hi all,
> >>
> >> I have just come across a weird state of operators after restore from
> >> checkpoint. After the restore, two operators that are connected (i.e.
> >> operator A is input of operator B) ended up with watermark of operator A
> >> being less than watermark of operator B. I don't know how to explain
> >> this. Can it be normal or does it signal a bug somewhere? If I
> >> understand Flink's checkpointing correctly, the checkpoint barrier flows
> >> from one operator to another, so the watermark should be aligned.
> >>
> >> I'm running a Beam pipeline on Flink 1.8.1.
> >>
> >> Am I missing something?
> >>
> >> Many thanks for comments,
> >>
> >>    Jan
> >>
> >>
>

[Jan Lukavský]

Actually, operator A is intermediate, source is preceding it.

On 8/7/19 2:44 PM, Kostas Kloudas wrote:

> Hi Jan,
>
> After looking at the code, my point 1) is false for *intermediate* tasks
> and if you are
> using a watermark assigner. This means that in these cases, Flink checks
> that the
> "next" watermark is greater than the "previous" one.
>
> But if your operator A is a source and you emit watermarks from the source,
> then
> it can happen that your watermark appears to go backwards on operator A,
> but
> operator B does the "correction" by discarding smaller watermarks. That can
> explain
> your observation.
>
> Cheers,
> Kostas
>
> On Wed, Aug 7, 2019 at 2:30 PM Jan Lukavský <[hidden email]> wrote:
>
>> Hi Kostas,
>>
>> thanks for reaction, comments inline.
>>
>> On 8/7/19 1:59 PM, Kostas Kloudas wrote:
>>> Hi Jan,
>>>
>>> Two pointers that may help you explain the behavior are the following:
>>>
>>> 1) If you have a custom watermark generator, I do not think that Flink
>>> checks if it emits only monotonically
>>> increasing watermarks. This is the responsibility of the generator
>> itself.
>>> This means that although you operator A
>>> is topologically before operator B, operator A may have a smaller
>> watermark
>>> if your watermark generator allows so.
>> I do generate watermarks by custom source, but I believe that the
>> generated sequence is monotonic. But still, I'm not sure, that even if
>> it was the case, that the generated watermark actually decreases, would
>> that mean, that downstream operator after source (operator A) would
>> actually "go back in time"?
>>> 2) Flink currently does not checkpoint the last seen watermark (
>>> https://issues.apache.org/jira/browse/FLINK-5601).
>>> This means that after restoring, your (event) time is assumed to be
>>> Long.Min until the first new watermark comes.
>>> So if you observed late data not being late anymore or sth similar, then
>> it
>>> may not be that the two operators have
>>> different watermarks but that after restoring event time rolls back to
>> the
>>> "beginning of time".
>> I actually didn't observe any wrong or unexpected behavior, exceptions
>> or wrong outputs. I just noticed this on Flink's WebUI and it looked
>> strange to me. Could it be just that the WebUI showed older watermark
>> for operator A? Strange was, that the watermarks were my screen long
>> enough to take a screenshot (so at least say 10 seconds displaying
>> watermark of operator A less than the one of operator B). Even if
>> watermarks are not checkpointed, would it still be possible for
>> watermark of operator B to be actually greater? I'm still confused of
>> how this could happen, because (in my understanding) output watermark of
>> operator A should be greater or equal to input watermark of B (because
>> it takes minimum of inputs).
>>
>> Sorry if I'm too digging into this, but I don't like things I cannot
>> explain, as they might point out to some bugs somewhere. :-) Or that my
>> mental model it not aligned with reality.
>>
>> Jan
>>
>>> I hope this helps,
>>> Kostas
>>>
>>> On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]> wrote:
>>>
>>>> Hi all,
>>>>
>>>> I have just come across a weird state of operators after restore from
>>>> checkpoint. After the restore, two operators that are connected (i.e.
>>>> operator A is input of operator B) ended up with watermark of operator A
>>>> being less than watermark of operator B. I don't know how to explain
>>>> this. Can it be normal or does it signal a bug somewhere? If I
>>>> understand Flink's checkpointing correctly, the checkpoint barrier flows
>>>> from one operator to another, so the watermark should be aligned.
>>>>
>>>> I'm running a Beam pipeline on Flink 1.8.1.
>>>>
>>>> Am I missing something?
>>>>
>>>> Many thanks for comments,
>>>>
>>>>     Jan
>>>>
>>>>

[Kostas Kloudas-4]

But are they chained together? Could you provide the code from your job, at
least until operator A?

On Wed, Aug 7, 2019 at 3:03 PM Jan Lukavský <[hidden email]> wrote:

> Actually, operator A is intermediate, source is preceding it.
>
> On 8/7/19 2:44 PM, Kostas Kloudas wrote:
> > Hi Jan,
> >
> > After looking at the code, my point 1) is false for *intermediate* tasks
> > and if you are
> > using a watermark assigner. This means that in these cases, Flink checks
> > that the
> > "next" watermark is greater than the "previous" one.
> >
> > But if your operator A is a source and you emit watermarks from the
> source,
> > then
> > it can happen that your watermark appears to go backwards on operator A,
> > but
> > operator B does the "correction" by discarding smaller watermarks. That
> can
> > explain
> > your observation.
> >
> > Cheers,
> > Kostas
> >
> > On Wed, Aug 7, 2019 at 2:30 PM Jan Lukavský <[hidden email]> wrote:
> >
> >> Hi Kostas,
> >>
> >> thanks for reaction, comments inline.
> >>
> >> On 8/7/19 1:59 PM, Kostas Kloudas wrote:
> >>> Hi Jan,
> >>>
> >>> Two pointers that may help you explain the behavior are the following:
> >>>
> >>> 1) If you have a custom watermark generator, I do not think that Flink
> >>> checks if it emits only monotonically
> >>> increasing watermarks. This is the responsibility of the generator
> >> itself.
> >>> This means that although you operator A
> >>> is topologically before operator B, operator A may have a smaller
> >> watermark
> >>> if your watermark generator allows so.
> >> I do generate watermarks by custom source, but I believe that the
> >> generated sequence is monotonic. But still, I'm not sure, that even if
> >> it was the case, that the generated watermark actually decreases, would
> >> that mean, that downstream operator after source (operator A) would
> >> actually "go back in time"?
> >>> 2) Flink currently does not checkpoint the last seen watermark (
> >>> https://issues.apache.org/jira/browse/FLINK-5601).
> >>> This means that after restoring, your (event) time is assumed to be
> >>> Long.Min until the first new watermark comes.
> >>> So if you observed late data not being late anymore or sth similar,
> then
> >> it
> >>> may not be that the two operators have
> >>> different watermarks but that after restoring event time rolls back to
> >> the
> >>> "beginning of time".
> >> I actually didn't observe any wrong or unexpected behavior, exceptions
> >> or wrong outputs. I just noticed this on Flink's WebUI and it looked
> >> strange to me. Could it be just that the WebUI showed older watermark
> >> for operator A? Strange was, that the watermarks were my screen long
> >> enough to take a screenshot (so at least say 10 seconds displaying
> >> watermark of operator A less than the one of operator B). Even if
> >> watermarks are not checkpointed, would it still be possible for
> >> watermark of operator B to be actually greater? I'm still confused of
> >> how this could happen, because (in my understanding) output watermark of
> >> operator A should be greater or equal to input watermark of B (because
> >> it takes minimum of inputs).
> >>
> >> Sorry if I'm too digging into this, but I don't like things I cannot
> >> explain, as they might point out to some bugs somewhere. :-) Or that my
> >> mental model it not aligned with reality.
> >>
> >> Jan
> >>
> >>> I hope this helps,
> >>> Kostas
> >>>
> >>> On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]> wrote:
> >>>
> >>>> Hi all,
> >>>>
> >>>> I have just come across a weird state of operators after restore from
> >>>> checkpoint. After the restore, two operators that are connected (i.e.
> >>>> operator A is input of operator B) ended up with watermark of
> operator A
> >>>> being less than watermark of operator B. I don't know how to explain
> >>>> this. Can it be normal or does it signal a bug somewhere? If I
> >>>> understand Flink's checkpointing correctly, the checkpoint barrier
> flows
> >>>> from one operator to another, so the watermark should be aligned.
> >>>>
> >>>> I'm running a Beam pipeline on Flink 1.8.1.
> >>>>
> >>>> Am I missing something?
> >>>>
> >>>> Many thanks for comments,
> >>>>
> >>>>     Jan
> >>>>
> >>>>
>

[Jan Lukavský]

Code would be a little complicated, because it is wrapped with several
layers of other APIs (Beam being one of them, but there is also other
layer).

I can provide the job graph [1]  a screenshot of the two watermarks [2].
The watermarks are taken from the two operators on bottom left.

Essentially, the job reads from Google cloud storage and simultaneously
from Kafka. On cloud storage are stored blob files containing historical
events and these blobs are marked with event time range (e.g. file is
named BLOB_EVENTS_TIMESTAMP1_TIMESTAMP2), and those timestamps are used
to generate watermarks from the batch storage (files are read in sorted
order).

Does that help, or would you like more details?

Jan

[1] https://transfer.sh/v473f/jobgraph.png
[2] https://transfer.sh/iDg1A/watermarks.png

On 8/7/19 3:04 PM, Kostas Kloudas wrote:

> But are they chained together? Could you provide the code from your job, at
> least until operator A?
>
> On Wed, Aug 7, 2019 at 3:03 PM Jan Lukavský <[hidden email]> wrote:
>
>> Actually, operator A is intermediate, source is preceding it.
>>
>> On 8/7/19 2:44 PM, Kostas Kloudas wrote:
>>> Hi Jan,
>>>
>>> After looking at the code, my point 1) is false for *intermediate* tasks
>>> and if you are
>>> using a watermark assigner. This means that in these cases, Flink checks
>>> that the
>>> "next" watermark is greater than the "previous" one.
>>>
>>> But if your operator A is a source and you emit watermarks from the
>> source,
>>> then
>>> it can happen that your watermark appears to go backwards on operator A,
>>> but
>>> operator B does the "correction" by discarding smaller watermarks. That
>> can
>>> explain
>>> your observation.
>>>
>>> Cheers,
>>> Kostas
>>>
>>> On Wed, Aug 7, 2019 at 2:30 PM Jan Lukavský <[hidden email]> wrote:
>>>
>>>> Hi Kostas,
>>>>
>>>> thanks for reaction, comments inline.
>>>>
>>>> On 8/7/19 1:59 PM, Kostas Kloudas wrote:
>>>>> Hi Jan,
>>>>>
>>>>> Two pointers that may help you explain the behavior are the following:
>>>>>
>>>>> 1) If you have a custom watermark generator, I do not think that Flink
>>>>> checks if it emits only monotonically
>>>>> increasing watermarks. This is the responsibility of the generator
>>>> itself.
>>>>> This means that although you operator A
>>>>> is topologically before operator B, operator A may have a smaller
>>>> watermark
>>>>> if your watermark generator allows so.
>>>> I do generate watermarks by custom source, but I believe that the
>>>> generated sequence is monotonic. But still, I'm not sure, that even if
>>>> it was the case, that the generated watermark actually decreases, would
>>>> that mean, that downstream operator after source (operator A) would
>>>> actually "go back in time"?
>>>>> 2) Flink currently does not checkpoint the last seen watermark (
>>>>> https://issues.apache.org/jira/browse/FLINK-5601).
>>>>> This means that after restoring, your (event) time is assumed to be
>>>>> Long.Min until the first new watermark comes.
>>>>> So if you observed late data not being late anymore or sth similar,
>> then
>>>> it
>>>>> may not be that the two operators have
>>>>> different watermarks but that after restoring event time rolls back to
>>>> the
>>>>> "beginning of time".
>>>> I actually didn't observe any wrong or unexpected behavior, exceptions
>>>> or wrong outputs. I just noticed this on Flink's WebUI and it looked
>>>> strange to me. Could it be just that the WebUI showed older watermark
>>>> for operator A? Strange was, that the watermarks were my screen long
>>>> enough to take a screenshot (so at least say 10 seconds displaying
>>>> watermark of operator A less than the one of operator B). Even if
>>>> watermarks are not checkpointed, would it still be possible for
>>>> watermark of operator B to be actually greater? I'm still confused of
>>>> how this could happen, because (in my understanding) output watermark of
>>>> operator A should be greater or equal to input watermark of B (because
>>>> it takes minimum of inputs).
>>>>
>>>> Sorry if I'm too digging into this, but I don't like things I cannot
>>>> explain, as they might point out to some bugs somewhere. :-) Or that my
>>>> mental model it not aligned with reality.
>>>>
>>>> Jan
>>>>
>>>>> I hope this helps,
>>>>> Kostas
>>>>>
>>>>> On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]> wrote:
>>>>>
>>>>>> Hi all,
>>>>>>
>>>>>> I have just come across a weird state of operators after restore from
>>>>>> checkpoint. After the restore, two operators that are connected (i.e.
>>>>>> operator A is input of operator B) ended up with watermark of
>> operator A
>>>>>> being less than watermark of operator B. I don't know how to explain
>>>>>> this. Can it be normal or does it signal a bug somewhere? If I
>>>>>> understand Flink's checkpointing correctly, the checkpoint barrier
>> flows
>>>>>> from one operator to another, so the watermark should be aligned.
>>>>>>
>>>>>> I'm running a Beam pipeline on Flink 1.8.1.
>>>>>>
>>>>>> Am I missing something?
>>>>>>
>>>>>> Many thanks for comments,
>>>>>>
>>>>>>      Jan
>>>>>>
>>>>>>

[Kostas Kloudas-4]

Hi Jan,

I am not sure what is happening. Operator A does not seem to be chained to
the source (which produces the watermarks) so
the check about increasing watermarks should be also applied there. BTW, I
assume that bottom left you mean the one that
starts with "activeDevices:takePresent..." (Op. A)  and
"activeDevices:stepLength..." (Op. B).

I am wondering if it can be that the WebUi is not consistent across
different operators.
For example, the watermark of Op B was simply not updated in the WebUI.

I also cc Chesnay who may have a better insight about the WebUi.

Cheers,
Kostas


On Wed, Aug 7, 2019 at 3:25 PM Jan Lukavský <[hidden email]> wrote:

> Code would be a little complicated, because it is wrapped with several
> layers of other APIs (Beam being one of them, but there is also other
> layer).
>
> I can provide the job graph [1]  a screenshot of the two watermarks [2].
> The watermarks are taken from the two operators on bottom left.
>
> Essentially, the job reads from Google cloud storage and simultaneously
> from Kafka. On cloud storage are stored blob files containing historical
> events and these blobs are marked with event time range (e.g. file is
> named BLOB_EVENTS_TIMESTAMP1_TIMESTAMP2), and those timestamps are used
> to generate watermarks from the batch storage (files are read in sorted
> order).
>
> Does that help, or would you like more details?
>
> Jan
>
> [1] https://transfer.sh/v473f/jobgraph.png
> [2] https://transfer.sh/iDg1A/watermarks.png
>
> On 8/7/19 3:04 PM, Kostas Kloudas wrote:
> > But are they chained together? Could you provide the code from your job,
> at
> > least until operator A?
> >
> > On Wed, Aug 7, 2019 at 3:03 PM Jan Lukavský <[hidden email]> wrote:
> >
> >> Actually, operator A is intermediate, source is preceding it.
> >>
> >> On 8/7/19 2:44 PM, Kostas Kloudas wrote:
> >>> Hi Jan,
> >>>
> >>> After looking at the code, my point 1) is false for *intermediate*
> tasks
> >>> and if you are
> >>> using a watermark assigner. This means that in these cases, Flink
> checks
> >>> that the
> >>> "next" watermark is greater than the "previous" one.
> >>>
> >>> But if your operator A is a source and you emit watermarks from the
> >> source,
> >>> then
> >>> it can happen that your watermark appears to go backwards on operator
> A,
> >>> but
> >>> operator B does the "correction" by discarding smaller watermarks. That
> >> can
> >>> explain
> >>> your observation.
> >>>
> >>> Cheers,
> >>> Kostas
> >>>
> >>> On Wed, Aug 7, 2019 at 2:30 PM Jan Lukavský <[hidden email]> wrote:
> >>>
> >>>> Hi Kostas,
> >>>>
> >>>> thanks for reaction, comments inline.
> >>>>
> >>>> On 8/7/19 1:59 PM, Kostas Kloudas wrote:
> >>>>> Hi Jan,
> >>>>>
> >>>>> Two pointers that may help you explain the behavior are the
> following:
> >>>>>
> >>>>> 1) If you have a custom watermark generator, I do not think that
> Flink
> >>>>> checks if it emits only monotonically
> >>>>> increasing watermarks. This is the responsibility of the generator
> >>>> itself.
> >>>>> This means that although you operator A
> >>>>> is topologically before operator B, operator A may have a smaller
> >>>> watermark
> >>>>> if your watermark generator allows so.
> >>>> I do generate watermarks by custom source, but I believe that the
> >>>> generated sequence is monotonic. But still, I'm not sure, that even if
> >>>> it was the case, that the generated watermark actually decreases,
> would
> >>>> that mean, that downstream operator after source (operator A) would
> >>>> actually "go back in time"?
> >>>>> 2) Flink currently does not checkpoint the last seen watermark (
> >>>>> https://issues.apache.org/jira/browse/FLINK-5601).
> >>>>> This means that after restoring, your (event) time is assumed to be
> >>>>> Long.Min until the first new watermark comes.
> >>>>> So if you observed late data not being late anymore or sth similar,
> >> then
> >>>> it
> >>>>> may not be that the two operators have
> >>>>> different watermarks but that after restoring event time rolls back
> to
> >>>> the
> >>>>> "beginning of time".
> >>>> I actually didn't observe any wrong or unexpected behavior, exceptions
> >>>> or wrong outputs. I just noticed this on Flink's WebUI and it looked
> >>>> strange to me. Could it be just that the WebUI showed older watermark
> >>>> for operator A? Strange was, that the watermarks were my screen long
> >>>> enough to take a screenshot (so at least say 10 seconds displaying
> >>>> watermark of operator A less than the one of operator B). Even if
> >>>> watermarks are not checkpointed, would it still be possible for
> >>>> watermark of operator B to be actually greater? I'm still confused of
> >>>> how this could happen, because (in my understanding) output watermark
> of
> >>>> operator A should be greater or equal to input watermark of B (because
> >>>> it takes minimum of inputs).
> >>>>
> >>>> Sorry if I'm too digging into this, but I don't like things I cannot
> >>>> explain, as they might point out to some bugs somewhere. :-) Or that
> my
> >>>> mental model it not aligned with reality.
> >>>>
> >>>> Jan
> >>>>
> >>>>> I hope this helps,
> >>>>> Kostas
> >>>>>
> >>>>> On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]>
> wrote:
> >>>>>
> >>>>>> Hi all,
> >>>>>>
> >>>>>> I have just come across a weird state of operators after restore
> from
> >>>>>> checkpoint. After the restore, two operators that are connected
> (i.e.
> >>>>>> operator A is input of operator B) ended up with watermark of
> >> operator A
> >>>>>> being less than watermark of operator B. I don't know how to explain
> >>>>>> this. Can it be normal or does it signal a bug somewhere? If I
> >>>>>> understand Flink's checkpointing correctly, the checkpoint barrier
> >> flows
> >>>>>> from one operator to another, so the watermark should be aligned.
> >>>>>>
> >>>>>> I'm running a Beam pipeline on Flink 1.8.1.
> >>>>>>
> >>>>>> Am I missing something?
> >>>>>>
> >>>>>> Many thanks for comments,
> >>>>>>
> >>>>>>      Jan
> >>>>>>
> >>>>>>
>

[Jan Lukavský]

Hi Kostas,

yes - operators A and B are as you said. The WebUI is what would explain
in best, if it is anyhow possible, that it would display sort of stale
information at one operator and updated on another.

Jan

On 8/7/19 3:59 PM, Kostas Kloudas wrote:

> Hi Jan,
>
> I am not sure what is happening. Operator A does not seem to be chained to
> the source (which produces the watermarks) so
> the check about increasing watermarks should be also applied there. BTW, I
> assume that bottom left you mean the one that
> starts with "activeDevices:takePresent..." (Op. A)  and
> "activeDevices:stepLength..." (Op. B).
>
> I am wondering if it can be that the WebUi is not consistent across
> different operators.
> For example, the watermark of Op B was simply not updated in the WebUI.
>
> I also cc Chesnay who may have a better insight about the WebUi.
>
> Cheers,
> Kostas
>
>
> On Wed, Aug 7, 2019 at 3:25 PM Jan Lukavský <[hidden email]> wrote:
>
>> Code would be a little complicated, because it is wrapped with several
>> layers of other APIs (Beam being one of them, but there is also other
>> layer).
>>
>> I can provide the job graph [1]  a screenshot of the two watermarks [2].
>> The watermarks are taken from the two operators on bottom left.
>>
>> Essentially, the job reads from Google cloud storage and simultaneously
>> from Kafka. On cloud storage are stored blob files containing historical
>> events and these blobs are marked with event time range (e.g. file is
>> named BLOB_EVENTS_TIMESTAMP1_TIMESTAMP2), and those timestamps are used
>> to generate watermarks from the batch storage (files are read in sorted
>> order).
>>
>> Does that help, or would you like more details?
>>
>> Jan
>>
>> [1] https://transfer.sh/v473f/jobgraph.png
>> [2] https://transfer.sh/iDg1A/watermarks.png
>>
>> On 8/7/19 3:04 PM, Kostas Kloudas wrote:
>>> But are they chained together? Could you provide the code from your job,
>> at
>>> least until operator A?
>>>
>>> On Wed, Aug 7, 2019 at 3:03 PM Jan Lukavský <[hidden email]> wrote:
>>>
>>>> Actually, operator A is intermediate, source is preceding it.
>>>>
>>>> On 8/7/19 2:44 PM, Kostas Kloudas wrote:
>>>>> Hi Jan,
>>>>>
>>>>> After looking at the code, my point 1) is false for *intermediate*
>> tasks
>>>>> and if you are
>>>>> using a watermark assigner. This means that in these cases, Flink
>> checks
>>>>> that the
>>>>> "next" watermark is greater than the "previous" one.
>>>>>
>>>>> But if your operator A is a source and you emit watermarks from the
>>>> source,
>>>>> then
>>>>> it can happen that your watermark appears to go backwards on operator
>> A,
>>>>> but
>>>>> operator B does the "correction" by discarding smaller watermarks. That
>>>> can
>>>>> explain
>>>>> your observation.
>>>>>
>>>>> Cheers,
>>>>> Kostas
>>>>>
>>>>> On Wed, Aug 7, 2019 at 2:30 PM Jan Lukavský <[hidden email]> wrote:
>>>>>
>>>>>> Hi Kostas,
>>>>>>
>>>>>> thanks for reaction, comments inline.
>>>>>>
>>>>>> On 8/7/19 1:59 PM, Kostas Kloudas wrote:
>>>>>>> Hi Jan,
>>>>>>>
>>>>>>> Two pointers that may help you explain the behavior are the
>> following:
>>>>>>> 1) If you have a custom watermark generator, I do not think that
>> Flink
>>>>>>> checks if it emits only monotonically
>>>>>>> increasing watermarks. This is the responsibility of the generator
>>>>>> itself.
>>>>>>> This means that although you operator A
>>>>>>> is topologically before operator B, operator A may have a smaller
>>>>>> watermark
>>>>>>> if your watermark generator allows so.
>>>>>> I do generate watermarks by custom source, but I believe that the
>>>>>> generated sequence is monotonic. But still, I'm not sure, that even if
>>>>>> it was the case, that the generated watermark actually decreases,
>> would
>>>>>> that mean, that downstream operator after source (operator A) would
>>>>>> actually "go back in time"?
>>>>>>> 2) Flink currently does not checkpoint the last seen watermark (
>>>>>>> https://issues.apache.org/jira/browse/FLINK-5601).
>>>>>>> This means that after restoring, your (event) time is assumed to be
>>>>>>> Long.Min until the first new watermark comes.
>>>>>>> So if you observed late data not being late anymore or sth similar,
>>>> then
>>>>>> it
>>>>>>> may not be that the two operators have
>>>>>>> different watermarks but that after restoring event time rolls back
>> to
>>>>>> the
>>>>>>> "beginning of time".
>>>>>> I actually didn't observe any wrong or unexpected behavior, exceptions
>>>>>> or wrong outputs. I just noticed this on Flink's WebUI and it looked
>>>>>> strange to me. Could it be just that the WebUI showed older watermark
>>>>>> for operator A? Strange was, that the watermarks were my screen long
>>>>>> enough to take a screenshot (so at least say 10 seconds displaying
>>>>>> watermark of operator A less than the one of operator B). Even if
>>>>>> watermarks are not checkpointed, would it still be possible for
>>>>>> watermark of operator B to be actually greater? I'm still confused of
>>>>>> how this could happen, because (in my understanding) output watermark
>> of
>>>>>> operator A should be greater or equal to input watermark of B (because
>>>>>> it takes minimum of inputs).
>>>>>>
>>>>>> Sorry if I'm too digging into this, but I don't like things I cannot
>>>>>> explain, as they might point out to some bugs somewhere. :-) Or that
>> my
>>>>>> mental model it not aligned with reality.
>>>>>>
>>>>>> Jan
>>>>>>
>>>>>>> I hope this helps,
>>>>>>> Kostas
>>>>>>>
>>>>>>> On Wed, Aug 7, 2019 at 12:11 PM Jan Lukavský <[hidden email]>
>> wrote:
>>>>>>>> Hi all,
>>>>>>>>
>>>>>>>> I have just come across a weird state of operators after restore
>> from
>>>>>>>> checkpoint. After the restore, two operators that are connected
>> (i.e.
>>>>>>>> operator A is input of operator B) ended up with watermark of
>>>> operator A
>>>>>>>> being less than watermark of operator B. I don't know how to explain
>>>>>>>> this. Can it be normal or does it signal a bug somewhere? If I
>>>>>>>> understand Flink's checkpointing correctly, the checkpoint barrier
>>>> flows
>>>>>>>> from one operator to another, so the watermark should be aligned.
>>>>>>>>
>>>>>>>> I'm running a Beam pipeline on Flink 1.8.1.
>>>>>>>>
>>>>>>>> Am I missing something?
>>>>>>>>
>>>>>>>> Many thanks for comments,
>>>>>>>>
>>>>>>>>       Jan
>>>>>>>>
>>>>>>>>