[FLINK-10050] Support 'allowedLateness' in CoGroupedStreams

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Implemented
• Affects Version/s: 1.5.1, 1.6.0
• Fix Version/s: 1.7.0
• Component/s: API / DataStream
• Labels: pull-request-available ready-to-commit windows
• Assignee: Unassigned
• Reporter: eugen yushin
• Created: 03/Aug/18 15:01
• Updated: 08/Jul/19 11:40
• Resolved: 21/Nov/18 17:01

Description

WindowedStream has a support of 'allowedLateness' feature, while CoGroupedStreams are not. At the mean time, WindowedStream is an inner part of CoGroupedStreams and all main functionality (like evictor/trigger/...) is simply delegated to WindowedStream.
There's no chance to operate with late arriving data from previous steps in cogroups (and joins). Consider the following flow:
a. read data from source1 -> aggregate data with allowed lateness
b. read data from source2 -> aggregate data with allowed lateness
c. cogroup/join streams a and b, and compare aggregated values

Step c doesn't accept any late data from steps a/b due to lack of `allowedLateness` API call in CoGroupedStreams.java.

Scope: add method `WithWindow.allowedLateness` to Java API (flink-streaming-java) and extend scala API (flink-streaming-scala).

Issue Links

• links to: GitHub Pull Request #6646