[FLINK-3646] Use Processing-Time Clock in Window Assigners/Triggers

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.0.0
• Component/s: API / DataStream
• Assignee: Kostas Kloudas
• Reporter: Aljoscha Krettek
• Created: 22/Mar/16 11:03
• Updated: 28/Feb/19 11:14
• Resolved: 30/Jun/16 21:33

Description

This is part of the effort to improve Window Triggers in Flink. We should replace all usage of System.currentTimeMillis() by a provided clock implementation. This way we can use a deterministic testing clock to verify the behavior of processing-time windowing components in unit tests.

This requires the following changes:

• Change StreamTask to have a Clock (name is WIP). By default this clock will use System.currentTimeMillis(). The clock must also provide an interface to register processing-time triggers, this is currently handled directly by StreamTask but must now also be handled by an external Clock
• Change API of WindowAssigner to take a context object that allows it to query the current processing time. This can be an abstract class AssignerContext with a single method long currentProcessingTime()
• Add a method long currentProcessingTime() in TriggerContext, change TriggerContext to use the clock methods provided by StreamTask and forwarded by WindowOperator
• Change processing-time triggers to use the new methods
• Change WindowOperator to support these changes

Issue Links

• is depended upon by: FLINK-3647 Change StreamSource to use Processing-Time Clock Service