Dawid Wysakowicz created FLINK-23473: ----------------------------------------
Summary: Do not create transaction in TwoPhaseCommitSinkFunction after finish() Key: FLINK-23473 URL: https://issues.apache.org/jira/browse/FLINK-23473 Project: Flink Issue Type: Sub-task Components: Connectors / Kafka, Runtime / Checkpointing Reporter: Dawid Wysakowicz In a scenario where: 1. task/operator received `finish()` 2. checkpoint 42 triggered (not yet completed) 3. checkpoint 43 triggered (not yet completed) 4. checkpoint 44 triggered (not yet completed) 5. notifyCheckpointComplete(43) And what should we do now? We can of course commit all transactions until checkpoint 43. But should we keep waiting for `notyifyCheckpointComplete(44)`? What if in the meantime another checkpoint is triggered? We could end up waiting indefinitely. Our proposal is to shutdown the task immediately after seeing first `notifyCheckpointComplete(X)`, where X is any triggered checkpoint AFTER `finish()`. This should be fine, as: a) ideally there should be no new pending transactions opened after checkpoint 42 b) even if operator/function is opening some transactions for checkpoint 43 and checkpoint 44 (`FlinkKafkaProducer`), those transactions after checkpoint 42 should be empty After seeing 5. (notifyCheckpointComplete(43)) It should be good enough to: - commit transactions from checkpoint 42, (and 43 if they were created, depends on the user code) - close operator, aborting any pending transactions (for checkpoint 44 if they were opened, depends on the user code) If checkpoint 44 completes afterwards, it will still be valid. Ideally we would recommend that after seeing `finish()` operators/functions should not be opening any new transactions. -- This message was sent by Atlassian Jira (v8.3.4#803005)