[
https://issues.apache.org/jira/browse/FLINK-39954?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18118770#comment-18118770
]
Martijn Visser commented on FLINK-39954:
----------------------------------------
Two more, one on master and one on release-1.20:
https://github.com/apache/flink/actions/runs/35599373328/job/106336599955
(master push 2026-09-21, JDK 17, parameters 1, 1)
https://github.com/apache/flink/actions/runs/35808755660/job/107022547202
(release-1.20 nightly 2026-09-23, Java 8, parameters 1, 2)
On master 82 commits were recorded instead of 80:
{code}
[ERROR]
org.apache.flink.test.streaming.runtime.SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling(int,
int, MiniCluster, ClusterClient)[3] -- Time elapsed: 2.621 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}
On release-1.20 the Hamcrest version of the same assertion reports an extra 662:
{code}
[ERROR]
org.apache.flink.test.streaming.runtime.SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling(int,
int, MiniCluster, ClusterClient)[1] -- Time elapsed: 3.024 s <<< FAILURE!
java.lang.AssertionError:
Expected: iterable over ["(895,null,-9223372036854775808)", ...] in any order
but: Not matched: "(662,null,-9223372036854775808)"
at
org.apache.flink.test.streaming.runtime.SinkV2ITCase.writerAndCommitterExecuteInStreamingModeWithScaling(SinkV2ITCase.java:253)
{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
> 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)
