[
https://issues.apache.org/jira/browse/BEAM-2393?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16404528#comment-16404528
]
Grzegorz Kołakowski commented on BEAM-2393:
-------------------------------------------
I'm afraid it does not. {{Thread.sleep(20)}} means "sleep for 20
*MILLI*seconds". I'm aware that {{sleep}} is in fact within synchronized
section.
Sorry for the miscommunication, I should have explained the problem more
precisely at the beginning. In a nutshell, for the following program:
{code:java}
p
.apply(GenerateSequence.from(0))
.apply(ParDo.of(new DoFn<Long, String>() {
@ProcessElement
public void processElement(ProcessContext context) {
log.info("Processing element {}", context.element());
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
context.output(context.element().toString());
}
}))
.apply(Window.into(FixedWindows.of(standardMinutes(1))));
{code}
I did snapshot ({{time bin/flink savepoint <job_id> <savepoints_path>}}) for
two versions of the Beam Flink Runner 10 times - with and without my changes
(lock scope extension, more precisely). I got the following results:
# Without my changes the command always takes similar time (from 1.192s to
1.271s).
# With my changes I obtained much longer times: 10.338s, 13.534s, 30.000s,
5.037s, 7.606s, 22.949s, 15.853s, 6.150s, 5.310s, 1.876s .
Without the sleep, the results for (1) and (2) are comparable. I understand
that 20ms for processing a single element is rather a rare case, but I just
thought it was worth mentioning every possible impact of my changes.
Should I be worried about this issue or should I ignore it for the time being?
> 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: Grzegorz Kołakowski
> 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)
