[FLINK-5420] Make CEP operators rescalable

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.2.0
• Component/s: Library / CEP
• Assignee: Kostas Kloudas
• Reporter: Kostas Kloudas
• Created: 06/Jan/17 13:38
• Updated: 28/Feb/17 18:53
• Resolved: 28/Feb/17 18:09

Description

This issue targets making the operators in the CEP library re-scalable. After this is implemented, the user will be able to take a savepoint and restart his job with a different parallelism.
This issue depends on https://issues.apache.org/jira/browse/FLINK-5845.

The way this is done is that we introduce the TimeServiceHandler in the AbstractStreamOperator, which keeps the registered InternalTimerService s (before this was in the AbstractStreamOperator) and a new service called KeyRegistry. The KeyRegistry will be fault tolerant and rescalable and will allow to register keys and a callback which will be invoked for each registered key upon reception of a watermark. This can be seen as keeping (recurring) timers for each of the registered keys that will fire "at the next watermark".

After introducing this service, upon reception of a watermark, all the processing of the NFAs will be delegated to the callback.

Issue Links

• links to: GitHub Pull Request #3307
• links to: GitHub Pull Request #3412

Sub-Tasks

• CEP: unify key and non-keyed operators: Closed
• CEP: make the operators backwards compatible.: Closed