[
https://issues.apache.org/jira/browse/FLINK-21133?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17285207#comment-17285207
]
Kezhu Wang commented on FLINK-21133:
------------------------------------
{quote}I think what stop-with-savepoint should really do is tell the mailbox to
take a savepoint and then exit (as if input was empty).
{quote}
Based on this, I would say that some of our assumptions and/or current
implementations are unnecessary or even harmful for future improvements:
* not interrupting threads, because don't need to be "as fast as possible". I
agree with this and I think this could be naming guarantee comparing to "cancel
--savepoint".
* calling close() on the operators for graceful shutdown and finalization. I
think this actually enforce "semantics guarantee" to "stop-with-savepoint" and
refactor bar for future improvements. After successful savepoint, it is nosense
to pretend "all records have been added to the operators" and "flush all
remaining buffered data". Just "cleanup resources" and exit is enough. Calling
close() also hurt implementation as we have to guarantee "flush all remaining
buffered data" to be success.
* Another data flow event(or flag of EndOfPartition). Assumed that "close" is
required, I think it is viable to replace "isStoppingBySyncSavepoint", but also
I think it is even more complicated than "isStoppingBySyncSavepoint" as it
requires supports from more than one stack: network channels(eg. InputGates),
stream tasks(eg. StreamTaskNetworkInput, StreamTask).
Given all above, I would say "stop-with-savepoint" as an utility method to
"take a savepoint and then exit" does not deserve any extra semantics
guarantees from streaming data flow comparing to "cancel --savepoint".
If we could reach above agreement, then I think the refactor is pretty simple,
just "take a savepoint, cleanup resources, then exit" in single task. No data
flow events will involving in. More importantly, we could clearly document that
"close" will only be "called after all records have been added to the
operators" to fix api semantics exception.
> FLIP-27 Source does not work with synchronous savepoint
> -------------------------------------------------------
>
> Key: FLINK-21133
> URL: https://issues.apache.org/jira/browse/FLINK-21133
> Project: Flink
> Issue Type: Bug
> Components: API / Core, API / DataStream, Runtime / Checkpointing
> Affects Versions: 1.11.3, 1.12.1
> Reporter: Kezhu Wang
> Priority: Critical
> Fix For: 1.11.4, 1.13.0, 1.12.3
>
>
> I have pushed branch
> [synchronous-savepoint-conflict-with-bounded-end-input-case|https://github.com/kezhuw/flink/commits/synchronous-savepoint-conflict-with-bounded-end-input-case]
> in my repository. {{SavepointITCase.testStopSavepointWithFlip27Source}}
> failed due to timeout.
> See also FLINK-21132 and
> [apache/iceberg#2033|https://github.com/apache/iceberg/issues/2033]..
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
