[hotfix] flink yarn-session submission mode bug fixed. (apache#820)

* flink remote submission mode bug fixed.

Author: wolfboys

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/entity/Application.java

@@ -312,6 +312,7 @@ public void setState(Integer state) {
      */
     public static Integer shouldTracking(@Nonnull FlinkAppState state) {
         switch (state) {
+            case ADDED:
             case CREATED:
             case FINISHED:
             case FAILED:

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/service/impl/ApplBuildPipeServiceImpl.java

@@ -44,6 +44,7 @@
 import com.streamxhub.streamx.console.core.enums.NoticeType;
 import com.streamxhub.streamx.console.core.enums.OptionState;
 import com.streamxhub.streamx.console.core.service.ApplicationBackUpService;
+import com.streamxhub.streamx.console.core.service.ApplicationConfigService;
 import com.streamxhub.streamx.console.core.service.AppBuildPipeService;
 import com.streamxhub.streamx.console.core.service.ApplicationService;
 import com.streamxhub.streamx.console.core.service.CommonService;
@@ -124,6 +125,9 @@ public class ApplBuildPipeServiceImpl
     @Autowired
     private ApplicationService applicationService;
 
+    @Autowired
+    private ApplicationConfigService applicationConfigService;
+
     private final ExecutorService executorService = new ThreadPoolExecutor(
             Runtime.getRuntime().availableProcessors() * 2,
             300,
@@ -239,6 +243,8 @@ public void onFinish(PipeSnapshot snapshot, BuildResult result) {
                     if (!app.isRunning()) {
                         if (app.isFlinkSqlJob()) {
                             applicationService.toEffective(app);
+                        } else {
+                            applicationConfigService.toEffective(app.getId(), app.getConfigId());
                         }
                     }
 

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/service/impl/ApplicationServiceImpl.java

@@ -32,6 +32,8 @@
 import com.streamxhub.streamx.common.enums.DevelopmentMode;
 import com.streamxhub.streamx.common.enums.ExecutionMode;
 import com.streamxhub.streamx.common.enums.ResolveOrder;
+import com.streamxhub.streamx.common.enums.StorageType;
+import com.streamxhub.streamx.common.fs.HdfsOperator;
 import com.streamxhub.streamx.common.util.DeflaterUtils;
 import com.streamxhub.streamx.common.util.ExceptionUtils;
 import com.streamxhub.streamx.common.util.ThreadUtils;
@@ -425,6 +427,12 @@ private void removeApp(Application application) {
         Long appId = application.getId();
         removeById(appId);
         application.getFsOperator().delete(application.getWorkspace().APP_WORKSPACE().concat("/").concat(appId.toString()));
+        try {
+            //曾经设置过yarn-application类型,尝试删除,不留后患.
+            HdfsOperator.delete(Workspace.of(StorageType.HDFS).APP_WORKSPACE().concat("/").concat(appId.toString()));
+        } catch (Exception e) {
+            //skip
+        }
     }
 
     @Override
@@ -812,6 +820,7 @@ public Application getApp(Application appParam) {
                 }
             }
         }
+
         if (ExecutionMode.YARN_SESSION.equals(application.getExecutionModeEnum())) {
             if (!application.getHotParamsMap().isEmpty()) {
                 if (application.getHotParamsMap().containsKey(ConfigConst.KEY_YARN_APP_ID())) {
@@ -896,7 +905,7 @@ public void cancel(Application appParam) {
                     String yarnQueue = (String) application.getHotParamsMap().get(ConfigConst.KEY_YARN_APP_QUEUE());
                     optionMap.put(ConfigConst.KEY_YARN_APP_QUEUE(), yarnQueue);
 
-                    if (!application.getHotParamsMap().isEmpty()) {
+                    if (ExecutionMode.YARN_SESSION.equals(application.getExecutionModeEnum())) {
                         String yarnSessionClusterId = (String) application.getHotParamsMap().get(ConfigConst.KEY_YARN_APP_ID());
                         assert yarnSessionClusterId != null;
                         extraParameter.put(ConfigConst.KEY_YARN_APP_ID(), yarnSessionClusterId);
@@ -1072,6 +1081,12 @@ public boolean start(Application appParam, boolean auto) throws Exception {
                 extraParameter.put(RestOptions.PORT.key(), activeAddress.getPort());
             }
 
+            if (ExecutionMode.YARN_SESSION.equals(application.getExecutionModeEnum())) {
+                String yarnSessionClusterId = (String) application.getHotParamsMap().get(ConfigConst.KEY_YARN_APP_ID());
+                assert yarnSessionClusterId != null;
+                extraParameter.put(ConfigConst.KEY_YARN_APP_ID(), yarnSessionClusterId);
+            }
+
             if (application.isFlinkSqlJob()) {
                 FlinkSql flinkSql = flinkSqlService.getEffective(application.getId(), false);
                 extraParameter.put(ConfigConst.KEY_FLINK_SQL(null), flinkSql.getSql());

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/service/impl/FlinkSqlServiceImpl.java

@@ -111,14 +111,14 @@ public List<FlinkSql> history(Application application) {
 
         List<FlinkSql> sqlList = this.baseMapper.selectList(wrapper);
         FlinkSql effective = getEffective(application.getId(), false);
-
-        for (FlinkSql sql : sqlList) {
-            if (sql.getId().equals(effective.getId())) {
-                sql.setEffective(true);
-                break;
+        if (effective != null && !sqlList.isEmpty()) {
+            for (FlinkSql sql : sqlList) {
+                if (sql.getId().equals(effective.getId())) {
+                    sql.setEffective(true);
+                    break;
+                }
             }
         }
-
         return sqlList;
     }
 

streamx-console/streamx-console-webapp/src/views/flink/app/View.vue

@@ -1904,17 +1904,9 @@ export default {
     handleEdit(app) {
       this.SetAppId(app.id)
       if (app.appType === 1) {
+        // jobType( 1 custom code 2: flinkSQL)
         this.$router.push({'path': '/flink/app/edit_streamx'})
-        if (app.jobType === 1) {
-          if (app.resourceForm === 1) {
-            this.$router.push({'path': '/flink/app/edit_streamx'})
-          } else {
-            this.$router.push({'path': '/flink/app/edit_flink'})
-          }
-        } else {
-          this.$router.push({'path': '/flink/app/edit_streamx'})
-        }
-      } else {
+      } else if (app.appType === 2) { //Apache Flink
         this.$router.push({'path': '/flink/app/edit_flink'})
       }
     },

streamx-plugin/streamx-flink-packer/src/main/scala/com/streamxhub/streamx/flink/packer/pipeline/BuildPipeline.scala

@@ -20,11 +20,12 @@
 package com.streamxhub.streamx.flink.packer.pipeline
 
 import com.streamxhub.streamx.common.util.{Logger, ThreadUtils}
+import com.streamxhub.streamx.flink.packer.pipeline.BuildPipeline.executor
 
-import java.util.concurrent.{LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit}
+import java.util.concurrent.{Callable, LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit}
 import scala.collection.JavaConverters._
 import scala.collection.mutable
-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService}
 import scala.util.{Failure, Success, Try}
 
 /**
@@ -161,7 +162,9 @@ trait BuildPipeline extends BuildPipelineProcess with BuildPipelineExpose with L
     Try {
       watcher.onStart(snapshot)
       logInfo(s"building pipeline is launching, params=${offerBuildParam.toString}")
-      buildProcess()
+      executor.submit(new Callable[BuildResult] {
+        override def call(): BuildResult = buildProcess()
+      }).get(3, TimeUnit.MINUTES)
     } match {
       case Success(result) =>
         pipeStatus = PipelineStatus.success
@@ -223,6 +226,6 @@ object BuildPipeline {
     new ThreadPoolExecutor.AbortPolicy
   )
 
-  implicit val executor: ExecutionContext = ExecutionContext.fromExecutorService(execPool)
+  implicit val executor: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(execPool)
 
 }