[
https://issues.apache.org/jira/browse/BEAM-2393?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16376528#comment-16376528
]
Grzegorz Kołakowski commented on BEAM-2393:
-------------------------------------------
I pushed my initial fix for BEAM-2393 to my
[fork|https://github.com/grzegorz8/beam/tree/BEAM-2393] (3 commits). I've used
{{BoundedToUnboundedSourceAdapter}} as suggested.
However, I still struggle with one issue. I've made an additional fix in
{{UnboundedSourceWrapper}}, which makes reader state update and element
emission atomic with respect to snapshot. Now the {{reader.advance()}} and
{{emitElement()}} are in the same synchronized section. It is necessary for two
reasons, Firstly, without the fix the snapshot is usually taken between
{{reader.advance()}} and {{emitElement(),}} which I suppose may lead to data
loss while recovering from such a snapshot. Secondly,
{{BoundedToUnboundedSourceAdapter.ResidualElements}} and
{{BoundedToUnboundedSourceAdapter.ResidualSource}} are requested for the next
element while their state is still uninitialized, which results in
NoSuchElementException.
Unfortunately, I'm afraid the fix affects snapshoting, namely, it seems to be
starving now. It takes a minute in average to obtain a savepoint:
{code:java}
$ time ./flink savepoint e6126b77abd47f5a3d3ea383608b4f53 <some_path>
Triggering savepoint for job e6126b77abd47f5a3d3ea383608b4f53.
Waiting for response...
Savepoint completed. Path: <some_path>
You can resume your program from this savepoint with the run command.
real 1m13.095s{code}
I'm afraid I'm not able to debug the problem on my own. I would appreciate any
help.
> BoundedSource is not fault-tolerant in FlinkRunner Streaming mode
> -----------------------------------------------------------------
>
> Key: BEAM-2393
> URL: https://issues.apache.org/jira/browse/BEAM-2393
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
> Reporter: Jingsong Lee
> Assignee: Jingsong Lee
> Priority: Major
>
> {{BoundedSourceWrapper}} does not implement snapshot() and restore(), when
> the failure to restart, it will send duplicate data.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
