Skip to content

[FLINK-40098][test-utils] Add FailingCheckpointedSource for exactly-once source tests#28686

Open
MartijnVisser wants to merge 2 commits into
apache:masterfrom
MartijnVisser:FLINK-40098
Open

[FLINK-40098][test-utils] Add FailingCheckpointedSource for exactly-once source tests#28686
MartijnVisser wants to merge 2 commits into
apache:masterfrom
MartijnVisser:FLINK-40098

Conversation

@MartijnVisser

@MartijnVisser MartijnVisser commented Jul 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

What is the purpose of the change

FLINK-32695 migrates test code off the deprecated SourceFunction API to Source V2 (FLIP-27). Roughly 15 checkpointing ITCases depend on a legacy FailingSource pattern: a source with a checkpointed position that fails exactly once after a completed checkpoint and resumes from the restored position. No Source V2 test utility offered these semantics.

This PR adds FailingCheckpointedSource to flink-test-utils-connector and migrates EventTimeAllWindowCheckpointingITCase as the first consumer. The remaining consumers follow in separate PRs under the umbrella.

Brief change log

  • Add FailingCheckpointedSource + SequenceSplit (a position-carrying split) to flink-test-utils-connector:
    • A snapshot past the arming threshold arms the failure; emission then pauses at the failure position; completion of the armed checkpoint triggers the throw; an aborted armed checkpoint re-arms on a later snapshot.
    • The reader holds emission just past the arming threshold until a snapshot arms, so recovery deterministically replays the arming-to-failure window regardless of emission speed or state backend (the legacy source relied on 1ms-per-element pacing for this; at full V2 speed the armed checkpoint would otherwise capture the failure position itself, losing detection power).
    • Recovery attempts are detected via the restored non-zero split position (SourceReaderContext exposes no attempt number).
    • The enumerator assigns one split per subtask only on first registration (a surviving enumerator on partial failover must not double-assign).
    • withSimulatedStateLossOnRecovery() exists purely so consuming tests can validate that they still detect exactly-once violations. Rescaling is unsupported (documented).
  • Migrate EventTimeAllWindowCheckpointingITCase off FailingSource. KeyedEventTimeGenerator (in the not-yet-migrated EventTimeWindowCheckpointingITCase) temporarily implements both generator interfaces so it can be shared; that file is touched only for this reason.

Verifying this change

  • 14 unit tests in FailingCheckpointedSourceTest cover the failure protocol, arming threshold, abort re-arm, restore-without-refailing, enumerator assignment/regression paths, and the serializer.
  • EventTimeAllWindowCheckpointingITCase passes 4/4.
  • Red check: with withSimulatedStateLossOnRecovery() temporarily wired in, testTumblingTimeWindow fails on duplicated window sums (e.g. [Window start: 1300 end: 1400] expected: 134950), proving the migrated test still detects exactly-once violations. Reverted afterwards.

Notes for reviewers

  • New classes are annotated @Experimental
  • flink-core's test-jar is added as a test-scope dependency of flink-test-utils-connector (for MockSplitEnumeratorContext).

Does this pull request potentially affect one of the following parts:

  • Dependencies (does it add or upgrade a dependency): no (test-scope only: flink-core test-jar)
  • The public API, i.e., is any changed class annotated with @Public(Evolving): no (new classes are @Experimental, test-scoped)
  • The serializers: no
  • The runtime per-record code paths (performance sensitive): no
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no (test-only)
  • The S3 file system connector: no

Documentation

  • Does this pull request introduce a new feature? no (test infrastructure)
  • If yes, how is the feature documented? not applicable
Was generative AI tooling used to co-author this PR?
  • Yes (Claude Fable 5)

Generated-by: Claude Fable 5

@MartijnVisser
[FLINK-40098][test-utils] Add FailingCheckpointedSource for exactly-o…
1857287 
…nce source tests

Replaces the fail-once-and-recover semantics of the legacy
SourceFunction-based FailingSource with a Source V2 utility

Generated-by: Claude Code (Fable 5)
@MartijnVisser
[FLINK-40098][tests] Migrate EventTimeAllWindowCheckpointingITCase to…
4915727 
… Source V2

Replaces the legacy FailingSource with FailingCheckpointedSource,
pinned to parallelism 1 like the implicitly non-parallel legacy source.
KeyedEventTimeGenerator implements both generator interfaces until the
remaining FailingSource consumers are migrated.

Generated-by: Claude Code (Fable 5)
@flinkbot

flinkbot commented Jul 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

Efrat19
Efrat19 reviewed Jul 12, 2026
View reviewed changes

@Efrat19 Efrat19 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks!
Generally LGTM % A little concern about the >500 lines here introduced to replace the 150-lines legacy FailingSource 😅 so most notes are focused on dead code for maintainability

Comment thread ...ils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
}

/** Determines whether and when the source fails. */
public static final class FailurePolicy implements Serializable {

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IMO FailurePolicy is a little bit of overengineering as we don't need FailingCheckpointedSource failure customization anywhere
and it also adds complexity bc a so called FailingCheckpointedSource doesn't guarantee failure.
I'd use FailingCheckpointedSource properties instead

Comment thread ...ils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
* Returns a copy of this source that stays idle after emitting all sequence numbers instead of
* finishing, for processing-time tests whose job is terminated externally.
*/
public FailingCheckpointedSource<T> withIdleAfterEmission() {

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

dead code

Comment thread ...ils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
* Only for validating that a test detects exactly-once violations; never use in a test's normal
* path.
*/
public FailingCheckpointedSource<T> withSimulatedStateLossOnRecovery() {

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

dead code

Comment thread ...ils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
@Nullable private ReaderOutput<T> delegate;

@Override
public void collect(T record) {

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

dead code

Comment thread ...ils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
private final FailurePolicy failurePolicy;
private final TypeInformation<T> producedType;
private final boolean idleAfterEmission;
private final boolean simulateStateLossOnRecovery;

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

idleAfterEmission and simulateStateLossOnRecovery flags are never actually turned on and only complicate FailingCheckpointedSourceReader logic, I'd introduce them along with their usage (which is probably never bc the legacy FailingSource also didn't support them)

Comment thread ...link-test-utils-connector/src/main/java/org/apache/flink/test/util/source/SequenceSplit.java
* parallelism.
*/
@Experimental
public class SequenceSplit implements SourceSplit {

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: move into FailingCheckpointedSource?

Comment thread ...link-test-utils-connector/src/main/java/org/apache/flink/test/util/source/SequenceSplit.java
* from the checkpointed position after recovery. The subtask index assumes a fixed source
* parallelism.
*/
@Experimental

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Make package-private instead of Experimental/public?

@github-actions github-actions Bot added the community-reviewed PR has been reviewed by the community. label Jul 12, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

1 more reviewer
@Efrat19 Efrat19 Efrat19 left review comments
Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

community-reviewed PR has been reviewed by the community.

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@MartijnVisser @flinkbot @Efrat19