[FLINK-21132] BoundedOneInput.endInput is called when taking synchronous savepoint

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.10.2, 1.10.3, 1.11.3, 1.12.1
• Fix Version/s: 1.11.4, 1.12.2, 1.13.0
• Component/s: Runtime / Task
• Labels: pull-request-available
• Release Note: endInput() is not called anymore (on BoundedOneInput and BoundedMultiInput) when the job is stopping with savepoint.
• Assignee: Roman Khachatryan
• Reporter: Kezhu Wang
• Created: 25/Jan/21 15:20
• Updated: 28/May/21 08:17
• Resolved: 03/Feb/21 19:58

Description

elkhand reported on project iceberg that BoundedOneInput.endInput was called when stopping job with savepoint.

I think it is a bug of Flink and was introduced in FLINK-14230. The changes rely on StreamTask.afterInvoke and OperatorChain.closeOperators will only be invoked after end of input. But that is not true long before after FLIP-34: Terminate/Suspend Job with Savepoint. Task could enter state called finished after synchronous savepoint, that is an expected job suspension and stopping.

sunhaibotb pnowojski roman_khachatryan Could you help confirm this ?

For full context, see apache/iceberg#2033. I have pushed branch synchronous-savepoint-conflict-with-bounded-end-input-case in my repository. Test case SavepointITCase.testStopSavepointWithBoundedInput failed due to BoundedOneInput.endInput called.

I am also aware of FLIP-147: Support Checkpoints After Tasks Finished, maybe the three should align on what finished means exactly. kkl0u chesnay gaoyunhaii

Issue Links

• causes: FLINK-21453 BoundedOneInput.endInput is NOT called when doing stop with savepoint WITH drain
• links to: GitHub Pull Request #14814
• links to: GitHub Pull Request #14815
• links to: GitHub Pull Request #14829