[
https://issues.apache.org/jira/browse/FLINK-3257?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15159353#comment-15159353
]
ASF GitHub Bot commented on FLINK-3257:
---------------------------------------
Github user senorcarbone commented on the pull request:
https://github.com/apache/flink/pull/1668#issuecomment-187825999
Thanks @tillrohrmann for the feedback!
I merged `ForwardingOneInputTask` into `StreamIterationSink`.
One more think I missed pointing out is that when the iteration timeout
occurs the `StreamIterationHead` flushes its log to its output before
terminating. An alternative take would be to delay termination until the
barrier arrives back to complete the snapshot. The problem with that version,
even though it's correct, is that under frequent checkpoint intervals the
iteration head could potentially never terminate because it will always be in
snapshotting state when the timeout occurs.
Regarding the state snapshotting, should I switch to using the ListState
representation instead or does it make no difference for the time being?
> Add Exactly-Once Processing Guarantees in Iterative DataStream Jobs
> -------------------------------------------------------------------
>
> Key: FLINK-3257
> URL: https://issues.apache.org/jira/browse/FLINK-3257
> Project: Flink
> Issue Type: Improvement
> Reporter: Paris Carbone
> Assignee: Paris Carbone
>
> The current snapshotting algorithm cannot support cycles in the execution
> graph. An alternative scheme can potentially include records in-transit
> through the back-edges of a cyclic execution graph (ABS [1]) to achieve the
> same guarantees.
> One straightforward implementation of ABS for cyclic graphs can work as
> follows along the lines:
> 1) Upon triggering a barrier in an IterationHead from the TaskManager start
> block output and start upstream backup of all records forwarded from the
> respective IterationSink.
> 2) The IterationSink should eventually forward the current snapshotting epoch
> barrier to the IterationSource.
> 3) Upon receiving a barrier from the IterationSink, the IterationSource
> should finalize the snapshot, unblock its output and emit all records
> in-transit in FIFO order and continue the usual execution.
> --
> Upon restart the IterationSource should emit all records from the injected
> snapshot first and then continue its usual execution.
> Several optimisations and slight variations can be potentially achieved but
> this can be the initial implementation take.
> [1] http://arxiv.org/abs/1506.08603
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
