[BEAM-2535] Allow explicit output time independent of firing specification for all timers

• Type: New Feature
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Fix Version/s: 2.7.1
• Component/s: beam-model, sdk-java-core
• Labels: Done stale-assigned
• Assignee: Reuven Lax
• Reporter: Kenneth Knowles
• Created: 28/Jun/17 19:56
• Updated: 06/Oct/20 21:56
• Resolved: 10/Jun/20 21:46

Description

Today, we have insufficient control over the event time timestamp of elements output from a timer callback.

1. For an event time timer, it is the timestamp of the timer itself.
2. For a processing time timer, it is the current input watermark at the time of processing.

But for both of these, we may want to reserve the right to output a particular time, aka set a "watermark hold".

A naive implementation of a TimerWithWatermarkHold would work for making sure output is not droppable, but does not fully explain window expiration and late data/timer dropping.

In the natural interpretation of a timer as a feedback loop on a transform, timers should be viewed as another channel of input, with a watermark, and items on that channel all need event time timestamps even if they are delivered according to a different time domain.

I propose that the specification for when a timer should fire should be separated (with nice defaults) from the specification of the event time of resulting outputs. These timestamps will determine a side channel with a new "timer watermark" that constrains the output watermark.

• We still need to fire event time timers according to the input watermark, so that event time timers fire.
• Late data dropping and window expiration will be in terms of the minimum of the input watermark and the timer watermark. In this way, whenever a timer is set, the window is not going to be garbage collected.
• We will need to make sure we have a way to "wake up" a window once it is expired; this may be as simple as exhausting the timer channel as soon as the input watermark indicates expiration of a window

This is mostly aimed at end-user timers in a stateful+timely DoFn. It seems reasonable to use timers as an implementation detail (e.g. in runners-core utilities) without wanting any of this additional machinery. For example, if there is no possibility of output from the timer callback.

Issue Links

• Blocked: BEAM-1589 Add OnWindowExpiration method to Stateful DoFn
• is depended upon by: BEAM-2925 Fn API user timer support
• relates to: BEAM-9566 Performance regression of FlinkRunner stream mode due to watermark holds update
• supercedes: BEAM-1696 Allow explicit output time for processing time timers
• supercedes: BEAM-1697 Allow output time that is some offset from target time of event time timer
• links to: GitHub Pull Request #4700
• links to: GitHub Pull Request #9582
• links to: GitHub Pull Request #10422
• links to: GitHub Pull Request #10627
• links to: GitHub Pull Request #10804