[FLINK-12351] AsyncWaitOperator should deep copy StreamElement when object reuse is enabled

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.14.0, 1.13.1
• Component/s: API / DataStream
• Labels: pull-request-available stale-assigned
• Assignee: Jark Wu
• Reporter: Jark Wu
• Created: 28/Apr/19 03:39
• Updated: 28/May/21 09:13
• Resolved: 24/Apr/21 13:58

Description

Currently, AsyncWaitOperator directly put the input StreamElement into StreamElementQueue. But when object reuse is enabled, the StreamElement is reused, which means the element in StreamElementQueue will be modified. As a result, the output of AsyncWaitOperator might be wrong.

An easy way to fix this might be deep copy the input StreamElement when object reuse is enabled, like this: https://github.com/apache/flink/blob/blink/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java#L209

Issue Links

• links to: GitHub Pull Request #8321