[
https://issues.apache.org/jira/browse/FLINK-39954?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18124953#comment-18124953
]
Martijn Visser commented on FLINK-39954:
----------------------------------------
Another one on master, push 2026-10-07 (JDK 17, parameters 1, 1), 82 commits
instead of 80 with the same two extra 895:
https://github.com/apache/flink/actions/runs/37606145603/job/112745650477
{code}
[ERROR]
org.apache.flink.test.streaming.runtime.SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling(int,
int, MiniCluster, ClusterClient)[3] -- Time elapsed: 2.560 s <<< FAILURE!
java.lang.AssertionError:
Expecting actual:
[Record{value=895, timestamp=null, watermark=-9223372036854775808}, ...
to contain exactly in any order:
[Record{value=895, timestamp=null, watermark=-9223372036854775808}, ...
but the following elements were unexpected:
[Record{value=895, timestamp=null, watermark=-9223372036854775808},
Record{value=895, timestamp=null, watermark=-9223372036854775808}]
at
org.apache.flink.test.streaming.runtime.SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling(SinkV2ITCase.java:201)
{code}
> SinkV2ITCase over-asserts commit multiplicity under unaligned checkpoints
> --------------------------------------------------------------------------
>
> Key: FLINK-39954
> URL: https://issues.apache.org/jira/browse/FLINK-39954
> Project: Flink
> Issue Type: Bug
> Components: API / Core
> Affects Versions: 1.20.6, 2.3.1, 2.4.0
> Reporter: Martijn Visser
> Priority: Critical
> Labels: pull-request-available, test-stability
>
> {{SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling}} asserts
> an exact commit multiplicity
> ({{containsExactlyInAnyOrder(duplicate(EXPECTED_COMMITTED_DATA_IN_STREAMING_MODE))}},
> i.e. each value committed exactly 2x) using a non-idempotent test committer
> ({{TrackingCommitter}} appends every {{commit()}} call). This is stronger
> than the {{Committer}} contract guarantees and makes the test fragile under
> unaligned checkpoints (UC), producing intermittent "unexpected element"
> (duplicate) failures.
> This surfaced while working on FLINK-38614
> I believe it's over-specified because the {{Committer}} contract
> ({{flink-core}} {{o.a.f.api.connector.sink2.Committer}} JavaDoc) guarantees
> an exactly-once effect and explicitly permits repeated commits: "A commit
> must be idempotent... Flink will restart from a previous checkpoint and
> re-attempt to commit all committables. Thus, some or all committables may
> have already been committed." It does not guarantee a specific number of
> {{commit()}} invocations.
> * Under aligned checkpoints the commit/re-commit pattern is deterministic
> (fixed barrier boundaries -> exactly the two-run replay), so
> {{duplicate(EXPECTED)}} holds.
> * Under unaligned checkpoints a boundary committable can shift, so it is
> committed in run 1, re-committed on run 2's restore, and committed again in
> run 2's normal processing -> 3x. That extra is contract-permitted (a real
> idempotent sink dedupes it to a no-op) but the non-idempotent counter records
> it as an "unexpected element".
> It looks like we can't easily make TrackingCommitter idempotent, because:
> * The committable is {{Record}} = (value, timestamp, watermark); under this
> test all records have {{timestamp=null, watermark=MIN}}, and the source emits
> each value twice legitimately, so those committables are byte-identical.
> There is no unique transaction/committable id to dedupe on ({{CommitRequest}}
> exposes only {{getCommittable()}} plus retry signals). Deduping by content
> would wrongly drop the legitimate second copy.
> * The test deliberately expects duplication ({{duplicate(EXPECTED)}}), so
> making the committer idempotent also requires changing the expected result to
> {{EXPECTED}}.
> A faithful fix gives the committable a unique identity (a
> {{TestSinkV2}}/{{Record}} plus source change), makes {{TrackingCommitter}}
> idempotent (dedupe by id, or use {{CommitRequest.signalAlreadyCommitted()}}),
> and asserts the exactly-once effect ({{EXPECTED}}, not
> {{duplicate(EXPECTED)}}). This is a non-trivial test redesign and should get
> sink/checkpointing maintainer review since it changes what the canonical
> SinkV2 test asserts.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
