Hanyu Zheng created FLINK-41217:
-----------------------------------
Summary: 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
Reporter: Hanyu Zheng
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)
