This is an automated email from the ASF dual-hosted git repository. dwysakowicz pushed a commit to branch release-1.14 in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push: new 59e3ec7 [FLINK-24355][runtime] Expose the flag for enabling checkpoints after tasks finish in the Web UI 59e3ec7 is described below commit 59e3ec7fcc95224c9c655949c2289f0702eac8a8 Author: SteNicholas <programg...@163.com> AuthorDate: Mon Oct 18 19:36:40 2021 +0800 [FLINK-24355][runtime] Expose the flag for enabling checkpoints after tasks finish in the Web UI --- .../shortcodes/generated/rest_v1_dispatcher.html | 3 +++ .../src/test/resources/rest_api_v1.snapshot | 3 +++ .../web-dashboard/src/app/interfaces/job-checkpoint.ts | 1 + .../job/checkpoints/job-checkpoints.component.html | 6 ++++++ .../job/checkpoints/CheckpointConfigHandler.java | 3 ++- .../rest/messages/checkpoints/CheckpointConfigInfo.java | 17 ++++++++++++++--- .../messages/checkpoints/CheckpointConfigInfoTest.java | 3 ++- 7 files changed, 31 insertions(+), 5 deletions(-) diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html index 88594fd..47428aa 100644 --- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html +++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html @@ -2237,6 +2237,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa }, "unaligned_checkpoints" : { "type" : "boolean" + }, + "checkpoints_after_tasks_finish" : { + "type" : "boolean" } } } </code> diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot index 7cacb30..c384e7d 100644 --- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot +++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot @@ -1275,6 +1275,9 @@ }, "aligned_checkpoint_timeout" : { "type" : "integer" + }, + "checkpoints_after_tasks_finish" : { + "type" : "boolean" } } } diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts index c91acd8..3371e20 100644 --- a/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts +++ b/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts @@ -125,6 +125,7 @@ export interface CheckPointConfigInterface { unaligned_checkpoints: boolean; tolerable_failed_checkpoints: number; aligned_checkpoint_timeout: number; + checkpoints_after_tasks_finish: boolean; } export interface CheckPointDetailInterface { diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html index 4c1b311..501c5c6 100644 --- a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html +++ b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html @@ -327,6 +327,12 @@ <td>Tolerable Failed Checkpoints</td> <td>{{ checkPointConfig['tolerable_failed_checkpoints'] }}</td> </tr> + <tr> + <td>Checkpoints With Finished Tasks</td> + <td> + {{ checkPointConfig['checkpoints_after_tasks_finish'] ? 'Enabled' : 'Disabled' }} + </td> + </tr> </ng-container> </tbody> </nz-table> diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java index 8ba71d9..f1868b0 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java @@ -131,7 +131,8 @@ public class CheckpointConfigHandler checkpointStorageName, checkpointCoordinatorConfiguration.isUnalignedCheckpointsEnabled(), checkpointCoordinatorConfiguration.getTolerableCheckpointFailureNumber(), - checkpointCoordinatorConfiguration.getAlignedCheckpointTimeout()); + checkpointCoordinatorConfiguration.getAlignedCheckpointTimeout(), + checkpointCoordinatorConfiguration.isEnableCheckpointsAfterTasksFinish()); } } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java index 56598d6..df6c88f 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java @@ -62,6 +62,9 @@ public class CheckpointConfigInfo implements ResponseBody { public static final String FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT = "aligned_checkpoint_timeout"; + public static final String FIELD_NAME_CHECKPOINTS_AFTER_TASKS_FINISH = + "checkpoints_after_tasks_finish"; + @JsonProperty(FIELD_NAME_PROCESSING_MODE) private final ProcessingMode processingMode; @@ -95,6 +98,9 @@ public class CheckpointConfigInfo implements ResponseBody { @JsonProperty(FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT) private final long alignedCheckpointTimeout; + @JsonProperty(FIELD_NAME_CHECKPOINTS_AFTER_TASKS_FINISH) + private final boolean checkpointsWithFinishedTasks; + @JsonCreator public CheckpointConfigInfo( @JsonProperty(FIELD_NAME_PROCESSING_MODE) ProcessingMode processingMode, @@ -108,7 +114,9 @@ public class CheckpointConfigInfo implements ResponseBody { @JsonProperty(FIELD_NAME_CHECKPOINT_STORAGE) String checkpointStorage, @JsonProperty(FIELD_NAME_UNALIGNED_CHECKPOINTS) boolean unalignedCheckpoints, @JsonProperty(FIELD_NAME_TOLERABLE_FAILED_CHECKPOINTS) int tolerableFailedCheckpoints, - @JsonProperty(FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT) long alignedCheckpointTimeout) { + @JsonProperty(FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT) long alignedCheckpointTimeout, + @JsonProperty(FIELD_NAME_CHECKPOINTS_AFTER_TASKS_FINISH) + boolean checkpointsWithFinishedTasks) { this.processingMode = Preconditions.checkNotNull(processingMode); this.checkpointInterval = checkpointInterval; this.checkpointTimeout = checkpointTimeout; @@ -120,6 +128,7 @@ public class CheckpointConfigInfo implements ResponseBody { this.unalignedCheckpoints = unalignedCheckpoints; this.tolerableFailedCheckpoints = tolerableFailedCheckpoints; this.alignedCheckpointTimeout = alignedCheckpointTimeout; + this.checkpointsWithFinishedTasks = checkpointsWithFinishedTasks; } @Override @@ -141,7 +150,8 @@ public class CheckpointConfigInfo implements ResponseBody { && Objects.equals(checkpointStorage, that.checkpointStorage) && unalignedCheckpoints == that.unalignedCheckpoints && tolerableFailedCheckpoints == that.tolerableFailedCheckpoints - && alignedCheckpointTimeout == that.alignedCheckpointTimeout; + && alignedCheckpointTimeout == that.alignedCheckpointTimeout + && checkpointsWithFinishedTasks == that.checkpointsWithFinishedTasks; } @Override @@ -157,7 +167,8 @@ public class CheckpointConfigInfo implements ResponseBody { checkpointStorage, unalignedCheckpoints, tolerableFailedCheckpoints, - alignedCheckpointTimeout); + alignedCheckpointTimeout, + checkpointsWithFinishedTasks); } /** Contains information about the externalized checkpoint configuration. */ diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfoTest.java index 9cdadc0..12911ff 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfoTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfoTest.java @@ -44,6 +44,7 @@ public class CheckpointConfigInfoTest "checkpointStorageName", true, 3, - 4); + 4, + true); } }