Adjust the priority of savepoint & Enhance create job client

Author: jefftlin

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/client/factory/AbstractJobClientFactory.scala

@@ -38,18 +38,7 @@ class AbstractJobClientFactory extends Logging {
     val client = getJobClientFactory(clientType)
       .createJobClient(onceJob, jobInfo, jobStateManager)
       .asInstanceOf[JobClient[LinkisJobInfo]]
-    Utils.tryThrow {
-      Utils.waitUntil(() => {
-        client.getJobInfo(true).asInstanceOf[EngineConnJobInfo].getApplicationId != null
-      }, Duration(10, TimeUnit.SECONDS), 100, 1000)
-      client
-    } {
-      case t: TimeoutException => {
-        logger.warn("Timeout to launch job, cannot get applicationId after deployment")
-        // Downgraded to yarn call
-        null
-      }
-    }
+    client
   }
 
   /**

streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/transform/impl/FlinkSavepointConfigTransform.scala

@@ -26,7 +26,7 @@ class FlinkSavepointConfigTransform extends FlinkConfigTransform {
         case (key, value) =>
           (FlinkConfigTransform.FLINK_CONFIG_PREFIX + key.replace(JobConfKeyConstants.SAVEPOINT.getValue, SAVE_POINT_PREFIX), value)
       }.asJava
-    Option(config.get(SAVE_POINT_PREFIX + "path")) match {
+    Option(config.get(FlinkConfigTransform.FLINK_CONFIG_PREFIX + SAVE_POINT_PREFIX + "path")) match {
       case Some(path) =>
         config.put(VAR_FLINK_SAVEPOINT_PATH.getValue, path)
       case _ =>