[FLINK-25695] TemporalJoin cause state leak in some cases

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.14.3
• Component/s: Table SQL / Runtime
• Labels: auto-deprioritized-major auto-deprioritized-minor pull-request-available
• Assignee: Unassigned
• Reporter: Lyn Zhang
• Created: 19/Jan/22 02:54
• Updated: 21/Aug/23 22:35

Description

Last year, I reported the similar bug of TemporalJoin cause state leak. Detail: FLINK-21833

Recently, I found the fix code can reduce the the leak size but can not resolve it completely.

The code of line 213 cause it and the right fix is to invoke cleanUp() method.

In FLINK-21833, we discussed when the code is running on line 213, that means Left State, Right State, registeredTimerState is empty, actually the Left State and Right State value(MapState) is empty but the key is still be in state, So invoke state.clear() is necessary. 

Issue Links

• Blocked: FLINK-21833 TemporalRowTimeJoinOperator.java will lead to the state expansion by short-life-cycle & huge RowData, although config idle.state.retention.time
• links to: GitHub Pull Request #18398