You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/04/03 15:16:04 UTC
[incubator-streampark] branch issue-2560 created (now 933ec1a0e)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a change to branch issue-2560
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
at 933ec1a0e [Feature] savepoint timeout support #2560
This branch includes the following new commits:
new 933ec1a0e [Feature] savepoint timeout support #2560
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-streampark] 01/01: [Feature] savepoint timeout support #2560
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch issue-2560
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
commit 933ec1a0e650f5feec0ce38d54c78f8427499ef0
Author: benjobs <be...@apache.org>
AuthorDate: Mon Apr 3 23:15:41 2023 +0800
[Feature] savepoint timeout support #2560
---
.../console/core/entity/Application.java | 1 +
.../core/service/impl/ApplicationServiceImpl.java | 1 +
.../src/api/flink/app/app.type.ts | 1 +
.../components/AppView/StopApplicationModal.vue | 24 ++++++++++++++++++----
.../flink/client/bean/CancelRequest.scala | 1 +
.../flink/client/trait/FlinkClientTrait.scala | 6 ++----
6 files changed, 26 insertions(+), 8 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
index 453220b53..38b072495 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
@@ -244,6 +244,7 @@ public class Application implements Serializable {
private transient String savePoint;
private transient Boolean savePointed = false;
private transient Boolean drain = false;
+ private transient Long savePointTimeout = 60L;
private transient Boolean allowNonRestored = false;
private transient String socketId;
private transient String projectName;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index 0246fc3b0..f63525e9c 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -1244,6 +1244,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
appParam.getSavePointed(),
appParam.getDrain(),
customSavepoint,
+ application.getSavePointTimeout(),
application.getK8sNamespace());
final Date triggerTime = new Date();
diff --git a/streampark-console/streampark-console-webapp/src/api/flink/app/app.type.ts b/streampark-console/streampark-console-webapp/src/api/flink/app/app.type.ts
index a56cb4dad..b5e2017e7 100644
--- a/streampark-console/streampark-console-webapp/src/api/flink/app/app.type.ts
+++ b/streampark-console/streampark-console-webapp/src/api/flink/app/app.type.ts
@@ -147,6 +147,7 @@ interface AppControl {
export interface CancelParam {
id: string;
savePointed: boolean;
+ savePointTimeout: number;
drain: boolean;
savePoint: string;
}
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StopApplicationModal.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StopApplicationModal.vue
index 12e8d77dc..d54cb4019 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StopApplicationModal.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StopApplicationModal.vue
@@ -23,6 +23,7 @@
</script>
<script setup lang="ts" name="StopApplicationModal">
+ import { InputNumber } from 'ant-design-vue';
import { BasicForm, useForm } from '/@/components/Form';
import { SvgIcon } from '/@/components/Icon';
import { BasicModal, useModalInner } from '/@/components/Modal';
@@ -68,6 +69,16 @@
afterItem: () => h('span', { class: 'conf-switch' }, 'cancel job with savepoint path'),
ifShow: ({ values }) => !!values.stopSavePointed,
},
+ {
+ field: 'savePointTimeout',
+ label: 'SavePoint Timeout',
+ component: 'Select',
+ afterItem: () => h('span', { class: 'conf-switch' }, 'savepoint timeout, Unit: second'),
+ slot: 'savePointTimeout',
+ defaultValue: 60,
+ ifShow: ({ values }) => !!values.stopSavePointed,
+ required: true,
+ },
{
field: 'drain',
label: 'Drain',
@@ -90,12 +101,14 @@
/* submit */
async function handleSubmit() {
try {
- const { stopSavePointed, drain, customSavepoint } = (await validate()) as Recordable;
+ const { stopSavePointed, customSavepoint, savePointTimeout, drain } =
+ (await validate()) as Recordable;
const stopReq = {
id: app.id,
savePointed: stopSavePointed,
- drain: drain,
savePoint: customSavepoint,
+ savePointTimeout: savePointTimeout,
+ drain: drain,
};
if (stopSavePointed) {
@@ -151,9 +164,12 @@
>
<template #title>
<SvgIcon name="shutdown" style="color: red" />
-
{{ t('flink.app.view.stop') }}
</template>
- <BasicForm @register="registerForm" class="!pt-20px" />
+ <BasicForm @register="registerForm" class="!pt-20px">
+ <template #savePointTimeout="{ model, field }">
+ <InputNumber v-model:value="model[field]" :min="10" :max="7200" step="10" />
+ </template>
+ </BasicForm>
</BasicModal>
</template>
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
index d6430fa8a..94acfd3c4 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
@@ -31,4 +31,5 @@ case class CancelRequest(flinkVersion: FlinkVersion,
override val withSavepoint: Boolean,
withDrain: Boolean,
savepointPath: String,
+ savePointTimeout: Long,
override val kubernetesNamespace: String = K8sFlinkConfig.DEFAULT_KUBERNETES_NAMESPACE) extends SavepointRequestTrait
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
index 0a4441f26..67a257294 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
@@ -475,16 +475,14 @@ trait FlinkClientTrait extends Logger {
val savePointDir: String = tryGetSavepointPathIfNeed(cancelRequest)
- val clientTimeout = getOptionFromDefaultFlinkConfig(cancelRequest.flinkVersion.flinkHome, ClientOptions.CLIENT_TIMEOUT)
-
val clientWrapper = new FlinkClusterClient(client)
(Try(cancelRequest.withSavepoint).getOrElse(false), Try(cancelRequest.withDrain).getOrElse(false)) match {
case (false, false) =>
client.cancel(jobID).get()
null
- case (true, false) => clientWrapper.cancelWithSavepoint(jobID, savePointDir).get(clientTimeout.toMillis, TimeUnit.MILLISECONDS)
- case (_, _) => clientWrapper.stopWithSavepoint(jobID, cancelRequest.withDrain, savePointDir).get(clientTimeout.toMillis, TimeUnit.MILLISECONDS)
+ case (true, false) => clientWrapper.cancelWithSavepoint(jobID, savePointDir).get(cancelRequest.savePointTimeout, TimeUnit.SECONDS)
+ case (_, _) => clientWrapper.stopWithSavepoint(jobID, cancelRequest.withDrain, savePointDir).get(cancelRequest.savePointTimeout, TimeUnit.SECONDS)
}
}