[
https://issues.apache.org/jira/browse/FLINK-21133?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17287735#comment-17287735
]
Kezhu Wang commented on FLINK-21133:
------------------------------------
[~becket_qin] [~pnowojski] [~sewen] Thanks all for recently discussions. It
helps me a lot in understanding:P.
[~sewen] I like the idea of clear-distinction between suspend and
pipeline-draining cases and "shut down the dataflow pipeline with one
checkpoint in total". The semantics of `StreamOperator.close` back to no
exception again in condition of clean definition of pipeline-draining case.
[~sewen] [~becket_qin] [~pnowojski] I am kind of unsure what is the difference
between pipeline-draining case and existing "terminate" case ? Where checkpoint
should happen in pipeline-draining case ? At trigger/barrier or after
end-of-stream-flush ? In case of trigger/barrier(eg. just like existing
"terminate" case), there is no big code-path difference comparing to suspend
case but clean definition. In case of after end-of-stream-flush, it is nearly
an real terminal operation, the savepoint may be hard to resume from. In "after
end-of-stream-flush" case, I think it is tight related to FLIP-147. In either
case, currently FLIP-143 sink does not work out with single checkpoint.
{quote}Maybe we would need to spin using
StreamTask#runSynchronousSavepointMailboxLoop while triggering the checkpoint,
and thus also blocking source thread from making any progress after triggering
the checkpoint?
{quote}
[~pnowojski] I think it is a must, otherwise if downstream closed before
source, {{SourceContext.collect}} will fail before {{notifyCheckpointComplete}}.
{quote}We would have to make sure that downstream/upstream task would cancel
correctly, without mis-leading error messages, if they receive network
connection closed before processing {{notifyCheckpointComplete()}}.
{quote}
[~pnowojski] Currently, {{RecordWriter}} reports error by lazy checking. I
think it make sense as network error may caused by caller module intentionally.
But I also saw exception swallowed in
{{PartitionRequestQueue.handleException}}. Anyway, it deserves more attentions.
Other issue about chained source in
{{MultipleInputStreamTask.triggerCheckpointAsync}}: {{advanceToEndOfEventTime}}
is ignored.
> 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)
