[
https://issues.apache.org/jira/browse/FLINK-41217?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hanyu Zheng updated FLINK-41217:
--------------------------------
Labels: 2.3.0 (was: 2.4)
> VALUES source emits its rows again when a streaming job restores from a
> checkpoint taken before the source exits
> ----------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41217
> URL: https://issues.apache.org/jira/browse/FLINK-41217
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Checkpointing, Table SQL / Planner
> Affects Versions: 2.3.0
> Reporter: Hanyu Zheng
> Priority: Major
> Labels: 2.3.0
>
> A VALUES clause in a streaming job is translated by CommonExecValues into an
> InputFormatSourceFunction. This source has no state: it does not record how
> many rows
> it has emitted. After a restore, the only thing that stops it from running
> again is the
> "fully finished" marker in the checkpoint (FLIP-147).
> That marker is not set as early as one would expect. After VALUES has emitted
> all of its
> rows, the task does not exit right away. It waits for one more checkpoint to
> complete
> (call it checkpoint N). Checkpoint N is triggered while the task is still
> RUNNING, so it
> stores no state for VALUES and does not mark it as finished. Only the next
> checkpoint
> (N+1), triggered after the task has exited, marks it as finished.
> If the job fails and is restored from checkpoint N (or any earlier
> checkpoint), VALUES
> reads all of its rows again from row 0. The downstream operators already
> counted these
> rows before the failure, so every row is processed twice. The window lasts
> from the job
> start until checkpoint N+1 completes, which is roughly one to two checkpoint
> intervals.
> Steps to reproduce:
> INSERT INTO snk
> SELECT id, COUNT(*) FROM (
> SELECT id FROM (VALUES (1), (2), (3)) AS v(id)
> UNION ALL SELECT 1000 AS id FROM gen -- unbounded datagen keeps the
> job alive
> ) GROUP BY id
> 1. Start the job with checkpointing enabled and wait ts rows.
> 2. Trigger a checkpoint while the VALUES vertex is still RUNNING
> (checkpoint N).
> The VALUES vertex becomes FINISHED after it comple
> 3. Cancel the job and restore it from checkpoint N.
> Expected: the sink holds (1, 1), (2, 1), (3, 1).
> Actual: the restored job reads the VALUES rows again,
> Restoring from checkpoint N+1 does not reproduce it, bes VALUES
> as finished.
> Proposed fix:
> Run VALUES with a source that checkpoints its position, so a restored job
> emits only the
> rows it has not emitted yet. Make the change only for Vnd leave
> all other InputFormatSourceFunction users untouched. Prefer the FLIP-27
> Source interface
> and reuse an existing source (for example DataGeneratorction,
> as StreamExecutionEnvironment#fromData does) instead of adding a new source.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
