Maximal watermark when two streams are connected

[Sung Gon Yi]

Hello,

Originally, watermark of connected stream is set by minimum of watermarks two streams when two streams are connected.
I wrote a code to connect two streams but one of streams does not have any message by a condition.
In this situation, watermark is never increased and processing is stuck.

I wonder there is an option to set a watermark as latest timestamp of either message.

Sincerely,
Sung Gon

[Jark Wu-3]

Hi Sung,

Watermark will be advanced only when records come in if you are using ".assignTimestampsAndWatermarks()". 

One way to solve this problem is you should call ".assignTimestampsAndWatermarks()" before the condition to make sure there are messages. 

Best,

Jark

On Thu, 22 Aug 2019 at 13:52, Sung Gon Yi <[hidden email]> wrote:

Hello,

Originally, watermark of connected stream is set by minimum of watermarks two streams when two streams are connected.
I wrote a code to connect two streams but one of streams does not have any message by a condition.
In this situation, watermark is never increased and processing is stuck.

I wonder there is an option to set a watermark as latest timestamp of either message.

Sincerely,
Sung Gon

[Sung Gon Yi]

I use assignerTimestampsAndWatermarks after connecting two streams and it works well.

Thank you.

On 22 Aug 2019, at 3:26 PM, Jark Wu <[hidden email]> wrote:

Hi Sung,

Watermark will be advanced only when records come in if you are using ".assignTimestampsAndWatermarks()". 

One way to solve this problem is you should call ".assignTimestampsAndWatermarks()" before the condition to make sure there are messages. 

Best,

Jark

On Thu, 22 Aug 2019 at 13:52, Sung Gon Yi <[hidden email]> wrote:

Hello,

Originally, watermark of connected stream is set by minimum of watermarks two streams when two streams are connected.
I wrote a code to connect two streams but one of streams does not have any message by a condition.
In this situation, watermark is never increased and processing is stuck.

I wonder there is an option to set a watermark as latest timestamp of either message.

Sincerely,
Sung Gon

[Fabian Hueske-2]

Hi Sung,

There is no switch to configure the WM to be the max of both streams and it would also in fact violate the core principles of the mechanism.

Watermarks are used to track the progress of event time in streams.

The implementations of operators rely on the fact that (almost) all records that are received in the future have a higher timestamp than the last received watermark.

If this does not hold true, many computations are performed too early and possibly many records are not included in the results.

If you have a stream that does not push any new data, you can set the sources to idle [1] and temporarily exclude them from the watermark tracking.

Best, Fabian

[1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/event_time.html#idling-sources

Am Do., 22. Aug. 2019 um 09:29 Uhr schrieb Sung Gon Yi <[hidden email]>:

I use assignerTimestampsAndWatermarks after connecting two streams and it works well.

Thank you.

On 22 Aug 2019, at 3:26 PM, Jark Wu <[hidden email]> wrote:

Hi Sung,

Watermark will be advanced only when records come in if you are using ".assignTimestampsAndWatermarks()". 

One way to solve this problem is you should call ".assignTimestampsAndWatermarks()" before the condition to make sure there are messages. 

Best,

Jark

On Thu, 22 Aug 2019 at 13:52, Sung Gon Yi <[hidden email]> wrote:

Hello,

Originally, watermark of connected stream is set by minimum of watermarks two streams when two streams are connected.
I wrote a code to connect two streams but one of streams does not have any message by a condition.
In this situation, watermark is never increased and processing is stuck.

I wonder there is an option to set a watermark as latest timestamp of either message.

Sincerely,
Sung Gon