[
https://issues.apache.org/jira/browse/FLINK-39954?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18123265#comment-18123265
]
Martijn Visser commented on FLINK-39954:
----------------------------------------
Two more on master, both duplicate commits:
https://github.com/apache/flink/actions/runs/37088407796/job/111106173675
(nightly 2026-10-03, Hadoop 3.4.3 profile, JDK 17, parameters 2, 1)
https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=79750&view=logs&j=8fd9202e-fd17-5b26-353c-ac1ff76c8f28
(nightly 2026-10-04, AdaptiveScheduler, JDK 17, parameters 1, 2)
The Azure one has the same two extra 895. The GitHub one recorded 90 commits
instead of 80:
{code}
[ERROR]
org.apache.flink.test.streaming.runtime.SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling(int,
int, MiniCluster, ClusterClient)[2] -- Time elapsed: 2.425 s <<< FAILURE!
java.lang.AssertionError:
Expecting actual:
[Record{value=127, 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=127, timestamp=null, watermark=-9223372036854775808},
Record{value=148, timestamp=null, watermark=-9223372036854775808},
Record{value=161, timestamp=null, watermark=-9223372036854775808},
Record{value=148, timestamp=null, watermark=-9223372036854775808},
Record{value=895, timestamp=null, watermark=-9223372036854775808},
Record{value=127, timestamp=null, watermark=-9223372036854775808},
Record{value=148, timestamp=null, watermark=-9223372036854775808},
Record{value=161, timestamp=null, watermark=-9223372036854775808},
Record{value=148, timestamp=null, watermark=-9223372036854775808}]
at
org.apache.flink.test.streaming.runtime.SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling(SinkV2ITCase.java:201)
{code}
With the release-2.3 one from 2026-10-02 that is three in the last seven days,
which makes this Critical rather than Major.
> 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: Major
> Labels: pull-request-available
>
> {{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)
