[Fix] submit an err job status alway is STARING

apache · Jan 9, 2024 · e1c386f · e1c386f
1 parent b537c1f
commit e1c386f
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 7 deletions.
diff --git a/...mpark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql b/...mpark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql
@@ -173,6 +173,7 @@ create table `t_flink_log` (
   `exception` text collate utf8mb4_general_ci,
   `option_time` datetime default null,
   `option_name` tinyint default null,
+  `user_id` bigint default null comment 'operator user id',
   primary key (`id`) using btree
 ) engine=innodb auto_increment=100000 default charset=utf8mb4 collate=utf8mb4_general_ci;
 

diff --git a/...apache/streampark/console/core/service/application/impl/ApplicationActionServiceImpl.java b/...apache/streampark/console/core/service/application/impl/ApplicationActionServiceImpl.java
@@ -498,7 +498,13 @@ public void start(Application appParam, boolean auto) throws Exception {
             String exception = ExceptionUtils.stringifyException(throwable);
             applicationLog.setException(exception);
             applicationLog.setSuccess(false);
-            applicationLogService.save(applicationLog);
+            try {
+              if (!applicationLogService.save(applicationLog)) {
+                log.error("save application log failed.");
+              }
+            } catch (Exception e) {
+              log.error("save application log failed.", e);
+            }
             if (throwable instanceof CancellationException) {
               doStopped(application);
             } else {
@@ -578,8 +584,7 @@ public void start(Application appParam, boolean auto) throws Exception {
    * @return
    */
   private boolean checkAppRepeatInYarn(String jobName) {
-    try {
-      YarnClient yarnClient = HadoopUtils.yarnClient();
+    try (YarnClient yarnClient = HadoopUtils.yarnClient()) {
       Set<String> types =
           Sets.newHashSet(
               ApplicationType.STREAMPARK_FLINK.getName(), ApplicationType.APACHE_FLINK.getName());
@@ -749,8 +754,7 @@ private Map<String, Object> getProperties(Application application) {
     }
 
     if (FlinkExecutionMode.isKubernetesApplicationMode(application.getExecutionMode())) {
-      try {
-        HadoopUtils.yarnClient();
+      try (YarnClient yarnClient = HadoopUtils.yarnClient()) {
         properties.put(JobManagerOptions.ARCHIVE_DIR.key(), Workspace.ARCHIVES_FILE_PATH());
       } catch (Exception e) {
         // skip
@@ -793,8 +797,9 @@ private void doStopped(Application appParam) {
         List<ApplicationReport> applications =
             applicationInfoService.getYarnAppReport(application.getJobName());
         if (!applications.isEmpty()) {
-          YarnClient yarnClient = HadoopUtils.yarnClient();
-          yarnClient.killApplication(applications.get(0).getApplicationId());
+          try (YarnClient yarnClient = HadoopUtils.yarnClient()) {
+            yarnClient.killApplication(applications.get(0).getApplicationId());
+          }
         }
       } catch (Exception ignored) {
       }