[FLINK-25256][streaming] Externally induced sources replay barriers received over RPC instead of inventing them out of thin air. #19138

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Merged

AHeise merged 3 commits into apache:master from AHeise:FLINK-25256

Mar 29, 2022

Contributor

AHeise commented Mar 17, 2022 •

edited

Loading

What is the purpose of the change

Externally induced sources are currently ill-defined and have a lot of unnecessary limitations. This PR addresses it by explicitly only holding back checkpoint barriers until the external source induces it.

While this approach seemingly limits the way externally induce sources work (they can't trigger a checkpoint on their own anymore), it actually explicitly supports the only plausible way. Sources simply can't trigger a checkpoint on their own - the checkpoint coordinator needs to track it and multiple sources need the coordinator to work at all (or else they deadlock).

Brief change log

Externally induced sources replay barriers received over RPC instead of inventing them out of thin air.
Clarify the contract.
Migrate related tests to JUnit5 and AssertJ.

Verifying this change

Expanded the original unit test to check that the barrier is correctly relayed.

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

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

Collaborator

flinkbot commented Mar 17, 2022 •

edited

Loading

CI report:

a64229d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

AHeise force-pushed the FLINK-25256 branch 2 times, most recently from 8ded29b to 99ed2d6 Compare

March 17, 2022 22:22

AHeise commented

View reviewed changes

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java Show resolved Hide resolved

flinkbot added the component=Runtime/Checkpointing label

AHeise force-pushed the FLINK-25256 branch from 99ed2d6 to 61c7489 Compare

March 18, 2022 08:52

AHeise commented

View reviewed changes

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java Outdated

Comment on lines 204 to 217

    
                          // note that at this point, we should probably not emit more data such that data is

                          // properly aligned

                          // however, unless we receive a reliable checkpoint abort RPC, this may deadlock

Contributor Author

AHeise Mar 18, 2022

We should probably discuss if this is the best choice.

AHeise commented

View reviewed changes

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java Outdated

Comment on lines 133 to 134

    
                                  // cleanup any old checkpoint that was cancelled before trigger

                                  triggeredCheckpoints.headSet(checkpointMetaData.getCheckpointId()).clear();

Contributor Author

AHeise Mar 18, 2022

The cleanup (and the one in #trigger) don't work well with concurrent checkpoints. Do we have a way to determine max concurrent checkpoints or can we actually rely on abortCheckpoint?

AHeise force-pushed the FLINK-25256 branch from 61c7489 to 628208d Compare

March 18, 2022 10:23

AHeise marked this pull request as ready for review

March 18, 2022 14:46

AHeise force-pushed the FLINK-25256 branch from 628208d to d3bff46 Compare

March 18, 2022 15:14

dawidwys reviewed

View reviewed changes

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java Outdated

    
                                  new UntriggeredCheckpoint(checkpointMetaData, checkpointOptions));

                          triggerFuture.complete(isRunning());

                      } else {

                          // not externally induced or trigger already received (rare case)

Contributor

dawidwys Mar 18, 2022

I guess, the comment is wrong now? It is only trigger already received (rare case), right?

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java Outdated Show resolved Hide resolved

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java Show resolved Hide resolved

dawidwys reviewed

View reviewed changes

Contributor

dawidwys left a comment

How hard would it be to add a test for the blocking/unblocking of the externally induced source?

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java Outdated Show resolved Hide resolved

...ng-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java

    
                      super.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);

                  /** Remove temporary data about a canceled checkpoint. */

                  private void cleanupCheckpoint(long checkpointId) {

                      assert (mailboxProcessor.isMailboxThread());

Contributor

dawidwys Mar 21, 2022

Shouldn't we potentially unblock the input here? If the only pending checkpoint was aborted/declined/cancelled?

Contributor Author

AHeise Mar 21, 2022

You are absolutely right.

AHeise force-pushed the FLINK-25256 branch 2 times, most recently from d9081db to f993e8f Compare

March 21, 2022 14:05

Contributor Author

AHeise commented Mar 21, 2022

How hard would it be to add a test for the blocking/unblocking of the externally induced source?

I have added assertions into the main test method that cover that. Please check if you think I should have additional test cases.

dawidwys approved these changes

View reviewed changes

Contributor

dawidwys left a comment

Looks fine to me now.

AHeise force-pushed the FLINK-25256 branch 3 times, most recently from 7a399ea to 5e263a6 Compare

March 23, 2022 12:37

AHeise mentioned this pull request

[FLINK-25256][streaming] Externally induced sources replay barriers received over RPC instead of inventing them out of thin air. [1.15] #19214

Merged

Contributor Author

AHeise commented Mar 23, 2022

This PR has been verified by the flink-pravega maintainers to work on their tests for checkpoints (savepoint test pending).

crazyzhou reviewed

View reviewed changes

Contributor

crazyzhou left a comment

I have tested with basic Pravega reader application savepointing locally with a rocksdb state backend, with a simple app
job graph:

After this fix, this app can successfully do the stop-with-savepoint while it failed before.

It can recover nicely with the savepoint:

The application is now only having _metadata file in each checkpoint and savepoint, so still trying some more complicated cases to see I can reproduce the issue.

Contributor

dawidwys commented Mar 24, 2022

On the matter of writing into separate files instead of keeping data inside of the metadata, you might want to have a look at: state.storage.fs.memory-threshold

crazyzhou approved these changes

View reviewed changes

Contributor

crazyzhou left a comment

I have tested with an app with a larger state, here is the savepoint structure:

:~/flink-savepoints/savepoint-9a7e45-0215c1d6f545$ ls
9c355659-2475-4402-a5e7-3450c70394b4  _metadata

The application with Pravega source can both cancel and stop with savepoint nicely, and can successfully recover from that.

Arvid Heise added 3 commits

March 29, 2022 10:10


          [refactor][streaming] Migrate Source(Operator)StreamTaskTest to JUnit…

fed0a1b

…5 and assertj


          [FLINK-25256][streaming] Externally induced sources replay barriers r…

1671baa

…eceived over RPC instead of inventing them out of thin air.

This change preserves the CheckpointOptions and properly integrates user-triggered snapshots and workflows with more than one source.
The externally induced source now merely delays the barrier instead of being able to insert one at a whim which would never work in aforementioned setups.


          [FLINK-25256][streaming] Clarify the contract of ExternallyInducedSou…

a64229d

…rce(Reader).

AHeise force-pushed the FLINK-25256 branch from 5e263a6 to a64229d Compare

March 29, 2022 08:13

AHeise mentioned this pull request

[FLINK-25256][streaming] Externally induced sources replay barriers received over RPC instead of inventing them out of thin air. [1.14] #19273

Closed

AHeise merged commit a4d194e into apache:master

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

component=Runtime/Checkpointing