[FLINK-30601] Omit "setKeyContextElement" call for non-keyed stream/operators to improve performance

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Done
• Fix Version/s: 1.17.0
• Component/s: Runtime / Task
• Labels: pull-request-available
• Assignee: Lijie Wang
• Reporter: Lijie Wang
• Created: 09/Jan/23 04:18
• Updated: 17/Jan/23 06:30
• Resolved: 17/Jan/23 06:30

Description

Currently, flink will set the correct key context(by call setKeyContextElement) before processing each record, which is typically used to extract key from record and pass that key to the state backends.

However, the "setKeyContextElement" is obviously not need for non-keyed stream/operator, in which case we can omit the "setKeyContextElement" calls to improve performance. Note that "setKeyContextElement" is an interface method, it requires looking up the interface table when calling, which will further increase the method call overhead.
 
We run the following program as benchmark with parallelism=1 and object re-use enabled. The benchmark results are averaged across 5 runs for each setup. Before and after applying the proposed change, the average execution time changed from 88.39 s to 78.76 s, which increases throughput by 10.8%.

env.fromSequence(1, 1000000000L)
        .map(x -> x)
        .map(x -> x)
        .map(x -> x)
        .map(x -> x)
        .map(x -> x).addSink(new DiscardingSink<>());

Issue Links

• links to: GitHub Pull Request #21621