[
https://issues.apache.org/jira/browse/FLINK-7883?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16216750#comment-16216750
]
Piotr Nowojski commented on FLINK-7883:
---------------------------------------
[~aljoscha] indeed it's not well suited, but it's only a part of the change
proposed by [~aphilippot]. The question is whether we want to block one on the
another or are we fine with some intermediate step. Introducing
{{StoppableFetchingSourceFunction}} (maybe not as a part of public api?) seems
like a simple extension over what we have now, which we could replace with
{{NewSource}} in the future.
Clean shutdown would be very helpful for Kafka's 0.11 exactly once producer -
without it we are leaving lingering transactions, that are blocking all of the
reads from Kafka topics (also by external systems) until Flink's application
will be restarted or until all of the transactions timeout.
> Stop fetching source before a cancel with savepoint
> ---------------------------------------------------
>
> Key: FLINK-7883
> URL: https://issues.apache.org/jira/browse/FLINK-7883
> Project: Flink
> Issue Type: Improvement
> Components: DataStream API, Kafka Connector, State Backends,
> Checkpointing
> Affects Versions: 1.4.0, 1.3.2
> Reporter: Antoine Philippot
>
> For a cancel with savepoint command, the JobManager trigger the cancel call
> once the savepoint is finished, but during the savepoint execution, kafka
> source continue to poll new messages which will not be part of the savepoint
> and will be replayed on the next application start.
> A solution could be to stop fetching the source stream task before triggering
> the savepoint.
> I suggest to add an interface {{StoppableFetchingSourceFunction}} with a
> method {{stopFetching}} that existant SourceFunction implementations could
> implement.
> We can add a {{stopFetchingSource}} property in
> {{CheckpointOptions}} class to pass the desired behaviour from
> {{JobManager.handleMessage(CancelJobWithSavepoint)}} to
> {{SourceStreamTask.triggerCheckpoint}}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
