[FLINK-32514] Support configuring checkpointing interval during process backlog #22931

yunfengzhou-hub · 2023-07-03T10:29:04Z

What is the purpose of the change

This pull request adds support that a Flink job can change its checkpointing interval depending on the process backlog status of the job.

Brief change log

Adds configuration execution.checkpointing.interval-during-backlog
Adds public method SplitEnumeratorContext#setIsProcessingBacklog
Makes HybridSource reports processing backlog depending on source stages
Supports changing checkpoint interval based on process backlog in CheckpointCoordinator

Verifying this change

This change added tests and can be verified as follows:

Added integration tests for changing checkpointing intervals.

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

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes
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
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs & JavaDocs

flinkbot · 2023-07-03T10:35:06Z

CI report:

e1d134f Azure: SUCCESS
b4ab880 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

yunfengzhou-hub · 2023-07-03T10:50:24Z

@flinkbot run azure

yunfengzhou-hub · 2023-07-03T10:50:40Z

@lindong28 could you please take a look at this PR?

lindong28

Thanks for the PR. Left some comments below.

...src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java

flink-core/src/main/java/org/apache/flink/api/connector/source/SplitEnumeratorContext.java

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java

...src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java

...treaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java

yunfengzhou-hub · 2023-07-20T05:55:30Z

@flinkbot run azure

yunfengzhou-hub · 2023-07-20T07:34:37Z

@flinkbot run azure

lindong28

Thanks for the update. Left some comments below.

...time/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java

...src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java

lindong28

Thanks for the update. Left some comments below.

...rc/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java

...ming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java

...rc/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java

lindong28

Thanks for the update. Left some comments below.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java

flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java

...src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java

flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java

yunfengzhou-hub · 2023-07-31T09:41:41Z

@flinkbot run azure

lindong28

Thanks for the update. Left some comments below.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java

...ntime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinator.java

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/VertexEndOfDataListener.java

lindong28

Thanks for the update. Left some comments below.

flink-runtime/src/main/java/org/apache/flink/runtime/source/event/EndOfDataEvent.java

...-core/src/test/java/org/apache/flink/util/concurrent/ManuallyTriggeredScheduledExecutor.java

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java

...-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDeciderTest.java

...src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java

...ests/src/test/java/org/apache/flink/test/util/NumberSequenceSourceBlockableByCheckpoint.java

docs/layouts/shortcodes/generated/execution_checkpointing_configuration.html

.../src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java

…g interval during process backlog

lindong28 · 2023-08-24T08:33:06Z

Thanks for the update. LGTM.

XComp · 2024-01-17T09:37:17Z

...src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java


-        context.lazyInitialize(globalFailureHandler, mainThreadExecutor);
+        context.lazyInitialize(globalFailureHandler, mainThreadExecutor, checkpointCoordinator);


@lindong28 @yunfengzhou-hub Is initializing the OperatorCoordinatorHolder twice done deliberately here? 🤔

@XComp Thanks for catching this. No, we didn't do it intentionally. I believe it is introduced when Yunfeng rebased the PR. I didn't catch this issue because I didn't go over the entire PR end-to-end very carefully when there is only minor remaining comments in the last 2 rounds of review.

It seems that the extra invocation of lazyInitialize() would not introduce any visible performance or correctness issue. Maybe one of us can fix it in our next PR.

Thanks for clarification. Doing it as a hotfix commit in one other PR makes sense because of the reasons you mentioned. 👍

yunfengzhou-hub force-pushed the FLINK-32514 branch 2 times, most recently from 57bf840 to 581e548 Compare July 3, 2023 10:49

yunfengzhou-hub marked this pull request as ready for review July 3, 2023 10:50

yunfengzhou-hub force-pushed the FLINK-32514 branch from 581e548 to 887666a Compare July 4, 2023 01:48

lindong28 reviewed Jul 18, 2023

View reviewed changes

yunfengzhou-hub force-pushed the FLINK-32514 branch 3 times, most recently from bc7f391 to 4fe5d1b Compare July 20, 2023 02:16

yunfengzhou-hub force-pushed the FLINK-32514 branch from 4fe5d1b to 5fc225f Compare July 20, 2023 08:20

lindong28 reviewed Jul 21, 2023

View reviewed changes

...time/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java Outdated Show resolved Hide resolved

...src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java Outdated Show resolved Hide resolved

yunfengzhou-hub force-pushed the FLINK-32514 branch 3 times, most recently from 17f9129 to 1039b03 Compare July 21, 2023 08:27

lindong28 reviewed Jul 23, 2023

View reviewed changes

yunfengzhou-hub force-pushed the FLINK-32514 branch 2 times, most recently from c521c4e to f0b9322 Compare July 24, 2023 08:22

lindong28 reviewed Jul 24, 2023

View reviewed changes

yunfengzhou-hub force-pushed the FLINK-32514 branch from f0b9322 to 2b10cd3 Compare July 24, 2023 10:34

lindong28 reviewed Jul 24, 2023

View reviewed changes

yunfengzhou-hub force-pushed the FLINK-32514 branch 6 times, most recently from 2dde6a8 to 3a28ae2 Compare July 31, 2023 05:42

lindong28 reviewed Jul 31, 2023

View reviewed changes

yunfengzhou-hub force-pushed the FLINK-32514 branch 4 times, most recently from 0725f05 to 10e471b Compare July 31, 2023 09:03

lindong28 reviewed Jul 31, 2023

View reviewed changes

yunfengzhou-hub force-pushed the FLINK-32514 branch 2 times, most recently from 18a7adc to d6c7573 Compare July 31, 2023 11:00

lindong28 reviewed Aug 1, 2023

View reviewed changes

yunfengzhou-hub force-pushed the FLINK-32514 branch 3 times, most recently from 91c04e6 to 5da3732 Compare August 2, 2023 00:42

lindong28 reviewed Aug 2, 2023

View reviewed changes

...ests/src/test/java/org/apache/flink/test/util/NumberSequenceSourceBlockableByCheckpoint.java Outdated Show resolved Hide resolved

yunfengzhou-hub force-pushed the FLINK-32514 branch from 5da3732 to 9c39634 Compare August 2, 2023 10:17

yunfengzhou-hub force-pushed the FLINK-32514 branch from 9c39634 to e1d134f Compare August 23, 2023 09:42

lindong28 reviewed Aug 24, 2023

View reviewed changes

docs/layouts/shortcodes/generated/execution_checkpointing_configuration.html Outdated Show resolved Hide resolved

.../src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java Outdated Show resolved Hide resolved

[FLINK-32514][Runtime/Checkpointing] Support configuring checkpointin…

b4ab880

…g interval during process backlog

yunfengzhou-hub force-pushed the FLINK-32514 branch from e1d134f to b4ab880 Compare August 24, 2023 07:00

lindong28 merged commit 615e824 into apache:master Aug 24, 2023

XComp reviewed Jan 17, 2024

View reviewed changes

yunfengzhou-hub mentioned this pull request Jan 19, 2024

[FLINK-34149][Runtime/Checkpointing] Fix SplitEnumeratorContext compatibility issue #24146

Merged

flinkbot added the component=Runtime/Checkpointing label Apr 4, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-32514] Support configuring checkpointing interval during process backlog #22931

[FLINK-32514] Support configuring checkpointing interval during process backlog #22931

yunfengzhou-hub commented Jul 3, 2023

flinkbot commented Jul 3, 2023 •

edited

yunfengzhou-hub commented Jul 3, 2023

yunfengzhou-hub commented Jul 3, 2023

lindong28 left a comment

yunfengzhou-hub commented Jul 20, 2023

yunfengzhou-hub commented Jul 20, 2023

lindong28 left a comment

lindong28 left a comment

lindong28 left a comment

yunfengzhou-hub commented Jul 31, 2023

lindong28 left a comment

lindong28 left a comment

lindong28 commented Aug 24, 2023

XComp Jan 17, 2024

lindong28 Jan 18, 2024 •

edited

XComp Jan 18, 2024


		context.lazyInitialize(globalFailureHandler, mainThreadExecutor);
		context.lazyInitialize(globalFailureHandler, mainThreadExecutor, checkpointCoordinator);

[FLINK-32514] Support configuring checkpointing interval during process backlog #22931

[FLINK-32514] Support configuring checkpointing interval during process backlog #22931

Conversation

yunfengzhou-hub commented Jul 3, 2023

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

flinkbot commented Jul 3, 2023 • edited

CI report:

yunfengzhou-hub commented Jul 3, 2023

yunfengzhou-hub commented Jul 3, 2023

lindong28 left a comment

Choose a reason for hiding this comment

yunfengzhou-hub commented Jul 20, 2023

yunfengzhou-hub commented Jul 20, 2023

lindong28 left a comment

Choose a reason for hiding this comment

lindong28 left a comment

Choose a reason for hiding this comment

lindong28 left a comment

Choose a reason for hiding this comment

yunfengzhou-hub commented Jul 31, 2023

lindong28 left a comment

Choose a reason for hiding this comment

lindong28 left a comment

Choose a reason for hiding this comment

lindong28 commented Aug 24, 2023

XComp Jan 17, 2024

Choose a reason for hiding this comment

lindong28 Jan 18, 2024 • edited

Choose a reason for hiding this comment

XComp Jan 18, 2024

Choose a reason for hiding this comment

flinkbot commented Jul 3, 2023 •

edited

lindong28 Jan 18, 2024 •

edited