You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by kr...@apache.org on 2023/04/06 10:33:43 UTC
[incubator-streampark] branch dev updated: [Bug] cancel job with savepoint bug fixed (#2577)
This is an automated email from the ASF dual-hosted git repository.
kriszu pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new ada8c3877 [Bug] cancel job with savepoint bug fixed (#2577)
ada8c3877 is described below
commit ada8c38779a7fff0fe1d8d1832ed306758a5c3fa
Author: benjobs <be...@apache.org>
AuthorDate: Thu Apr 6 18:33:37 2023 +0800
[Bug] cancel job with savepoint bug fixed (#2577)
Co-authored-by: benjobs <be...@gmail.com>
---
.../streampark/console/core/entity/YarnQueue.java | 1 -
.../core/service/impl/ApplicationServiceImpl.java | 2 +-
.../flink/client/trait/FlinkClientTrait.scala | 21 +++++++++++----------
3 files changed, 12 insertions(+), 12 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/YarnQueue.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/YarnQueue.java
index 35d30effb..b6e5e4bd4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/YarnQueue.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/YarnQueue.java
@@ -34,7 +34,6 @@ import java.util.Date;
public class YarnQueue implements Serializable {
@TableId(type = IdType.AUTO)
- @TableField(fill = FieldFill.INSERT, updateStrategy = FieldStrategy.NEVER)
private Long id;
@TableField(fill = FieldFill.INSERT, updateStrategy = FieldStrategy.NEVER)
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 6cdc10543..3fea528b4 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
@@ -1250,7 +1250,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
appParam.getSavePointed(),
appParam.getDrain(),
customSavepoint,
- application.getSavePointTimeout(),
+ appParam.getSavePointTimeout(),
application.getK8sNamespace());
final Date triggerTime = new Date();
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 0b981a6fa..ef4459bf1 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
@@ -120,7 +120,7 @@ trait FlinkClientTrait extends Logger {
}
- def setJvmOptions(submitRequest: SubmitRequest, flinkConfig: Configuration): Unit = {
+ private[this] def setJvmOptions(submitRequest: SubmitRequest, flinkConfig: Configuration): Unit = {
if (MapUtils.isNotEmpty(submitRequest.properties)) {
submitRequest.properties.foreach(x => {
val k = x._1.trim
@@ -165,15 +165,16 @@ trait FlinkClientTrait extends Logger {
logInfo(
s"""
|----------------------------------------- flink job cancel --------------------------------
- | userFlinkHome : ${cancelRequest.flinkVersion.flinkHome}
- | flinkVersion : ${cancelRequest.flinkVersion.version}
- | clusterId : ${cancelRequest.clusterId}
- | withSavePoint : ${cancelRequest.withSavepoint}
- | savePointPath : ${cancelRequest.savepointPath}
- | withDrain : ${cancelRequest.withDrain}
- | k8sNamespace : ${cancelRequest.kubernetesNamespace}
- | appId : ${cancelRequest.clusterId}
- | jobId : ${cancelRequest.jobId}
+ | userFlinkHome : ${cancelRequest.flinkVersion.flinkHome}
+ | flinkVersion : ${cancelRequest.flinkVersion.version}
+ | clusterId : ${cancelRequest.clusterId}
+ | withSavePoint : ${cancelRequest.withSavepoint}
+ | savePointPath : ${cancelRequest.savepointPath}
+ | savePointTimeout : ${cancelRequest.savePointTimeout}
+ | withDrain : ${cancelRequest.withDrain}
+ | k8sNamespace : ${cancelRequest.kubernetesNamespace}
+ | appId : ${cancelRequest.clusterId}
+ | jobId : ${cancelRequest.jobId}
|-------------------------------------------------------------------------------------------
|""".stripMargin)
val flinkConf = new Configuration()