[jira] [Comment Edited] (FLINK-30238) Unified Sink committer does not clean up state on final savepoint

["Tzu-Li (Gordon) Tai (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-30238?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17761051#comment-17761051 ] 

Tzu-Li (Gordon) Tai edited comment on FLINK-30238 at 8/31/23 5:39 PM:
----------------------------------------------------------------------

hi all, I'd like to move this ticket forward and make a call if things are working as expected, or there's an actual bug.

First of all, lets narrow down the scope of this ticket to address only the following :
{{stop-with-savepoint}} stores last CommittableSummary with LONG_MAX checkpoint ID to be persisted in savepoints, which prevents future {{stop-with-savepoint}} to succeed after restore because CommittableSummaries should always have different checkpoint IDs.

The separate issue with post-commit topologies not receiving committed committables can be addressed separately.

Summarizing the facts we have so far:
 * If a LONG_MAX checkpoint ID special marker is propagated, it means {{endOfInput}} was called and it is expected behavior that the job will not function properly if restored from the generated savepoint. {{endOfInput}} should only ever be called when 1) reaching bounded end of input, or 2) stop-with-savepoint with the --drain option.
 * The reason why the Kafka ITCase referenced by [~fpaul] has {{endOfInput}} being called is because of the setup {{{}advanceTimestamp = true{}}}, which translates equivalently to stop-with-savepoint with --drain.

With the above, can we conclude that:
 * {{stop-with-savepoint}} without draining works as expected, such that {{endOfInput}} is not being called and the LONG_MAX CommittableSummary is not being sent and stored as committables state.
 * {{stop-with-savepoint}} _with draining_ is also working as expected - jobs restored from these savepoints are not expected to function properly in the first place.

 

I'm looking to verify that {{stop-with-savepoint}} without draining indeed doesn't call the {{endOfInput}} method - if that's the case, then I think we have a case closed? If that's the case - the only fix we need is to adjust the Kafka ITCase.

In parallel, just wanna jot these thoughts down to see if I'm missing anything obvious cc [~fpaul] [~gaoyunhaii].


was (Author: tzulitai):
hi all, I'd like to move this ticket forward and make a call if things are working as expected, or there's an actual bug.

First of all, lets narrow down the scope of this ticket to address only the following :
{{stop-with-savepoint}} stores last CommittableSummary with LONG_MAX checkpoint ID to be persisted in savepoints, which prevents future {{stop-with-savepoint}} to succeed after restore because CommittableSummaries should always have different checkpoint IDs.

The separate issue with post-commit topologies not receiving committed committables can be addressed separately.

Summarizing the facts we have so far:
 * If a LONG_MAX checkpoint ID special marker is propagated, it means {{endOfInput}} was called and it is expected behavior that the job will not function properly if restored from the generated savepoint. {{endOfInput}} should only ever be called when 1) reaching bounded end of input, or 2) stop-with-savepoint with the --drain option.
 * The reason why the Kafka ITCase referenced by [~fpaul] has {{endOfInput}} being called is because of the setup {{{}advanceTimestamp = true{}}}, which translates equivalently to stop-with-savepoint with --drain.


With the above, can we conclude that:
 * {{stop-with-savepoint}} without draining works as expected, such that {{endOfInput}} is not being called and the LONG_MAX CommittableSummary is not being sent and stored as committables state.
 * {{stop-with-savepoint}} _with draining_ is also working as expected - jobs restored from these savepoints are not expected to function properly in the first place.

 

I'm looking to verify that {{stop-with-savepoint}} without draining indeed doesn't call the {{endOfInput}} method - if that's the case, then I think we have a case closed? In parallel, just wanna jot these thoughts down to see if I'm missing anything obvious cc [~fpaul] [~gaoyunhaii].

> Unified Sink committer does not clean up state on final savepoint
> -----------------------------------------------------------------
>
>                 Key: FLINK-30238
>                 URL: https://issues.apache.org/jira/browse/FLINK-30238
>             Project: Flink
>          Issue Type: Bug
>          Components: Connectors / Common
>    Affects Versions: 1.17.0, 1.15.3, 1.16.1
>            Reporter: Fabian Paul
>            Priority: Critical
>         Attachments: Screenshot 2023-03-09 at 1.47.11 PM.png, image (8).png
>
>
> During stop-with-savepoint the committer only commits the pending committables on notifyCheckpointComplete.
> This has several downsides.
>  * Last committableSummary has checkpoint id LONG.MAX and is never cleared from the state leading to that stop-with-savepoint does not work when the pipeline recovers from a savepoint 
>  * While the committables are committed during stop-with-savepoint they are not forwarded to post-commit topology, potentially losing data and preventing to close open transactions.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)