[FLINK-21421] Add coMapWithState to ConnectedStreams

• Type: New Feature
• Status: Closed
• Priority: Minor
• Resolution: Won't Do
• Component/s: API / Scala
• Assignee: Unassigned
• Reporter: Aris Koliopoulos
• Created: 19/Feb/21 17:06
• Updated: 28/Oct/21 16:07
• Resolved: 28/Oct/21 16:07

Description

Currently there is no syntactic sugar for stateful functions in `ConnectedStreams` in Scala. 

This makes stateful joins (aka `connect`) more verbose and exposes users to Java interfaces (by requiring a `RichCoMapFunction` implementation to access state in `ConnectedStreams`).

Looking at DriveTribe's codebase, we have implemented ~80% of our ConnectedStreams operators using this `coMapWithState` implementation:

https://github.com/ariskk/flink-stream-join/blob/main/src/main/scala/com/ariskk/streamjoin/ConnectedStreamsOps.scala#L15

A `coFlatMapWithState` can be trivially implemented on top.

This has been in production for so long I forgot it was our code and not Flink's.

I can easily add it if this is of interest. 

I did the work, here is the diff https://github.com/apache/flink/compare/master...ariskk:FLINK-21421