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);
}
}
