[Improve] spark app submit bug fixed.

qodana.yaml

@@ -0,0 +1,47 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#-------------------------------------------------------------------------------#
+#               Qodana analysis is configured by qodana.yaml file               #
+#             https://www.jetbrains.com/help/qodana/qodana-yaml.html            #
+#-------------------------------------------------------------------------------#
+version: "1.0"
+
+#Specify inspection profile for code analysis
+profile:
+  name: qodana.starter
+
+#Enable inspections
+#include:
+#  - name: <SomeEnabledInspectionId>
+
+#Disable inspections
+#exclude:
+#  - name: <SomeDisabledInspectionId>
+#    paths:
+#      - <path/where/not/run/inspection>
+
+projectJDK: 8 #(Applied in CI/CD pipeline)
+
+#Execute shell command before Qodana execution (Applied in CI/CD pipeline)
+#bootstrap: sh ./prepare-qodana.sh
+
+#Install IDE plugins before Qodana execution (Applied in CI/CD pipeline)
+#plugins:
+#  - id: <plugin.id> #(plugin id can be found at https://plugins.jetbrains.com)
+
+#Specify Qodana linter for analysis (Applied in CI/CD pipeline)
+linter: jetbrains/qodana-jvm:latest

streampark-common/src/main/scala/org/apache/streampark/common/util/PropertiesUtils.scala

@@ -37,12 +37,6 @@ object PropertiesUtils extends Logger {
 
   private[this] lazy val PROPERTY_PATTERN = Pattern.compile("(.*?)=(.*?)")
 
-  private[this] lazy val SPARK_PROPERTY_COMPLEX_PATTERN = Pattern.compile("^[\"']?(.*?)=(.*?)[\"']?$")
-
-  // scalastyle:off
-  private[this] lazy val SPARK_ARGUMENT_REGEXP = "\"?(\\s+|$)(?=(([^\"]*\"){2})*[^\"]*$)\"?"
-  // scalastyle:on
-
   private[this] lazy val MULTI_PROPERTY_REGEXP = "-D(.*?)\\s*=\\s*[\\\"|'](.*)[\\\"|']"
 
   private[this] lazy val MULTI_PROPERTY_PATTERN = Pattern.compile(MULTI_PROPERTY_REGEXP)
@@ -398,7 +392,7 @@ object PropertiesUtils extends Logger {
         case d if Utils.isNotEmpty(d) =>
           d.foreach(x => {
             if (x.nonEmpty) {
-              val p = SPARK_PROPERTY_COMPLEX_PATTERN.matcher(x)
+              val p = PROPERTY_PATTERN.matcher(x)
               if (p.matches) {
                 map += p.group(1).trim -> p.group(2).trim
               }
@@ -409,22 +403,4 @@ object PropertiesUtils extends Logger {
       map.toMap
     }
   }
-
-  /** extract spark configuration from sparkApplication.appArgs */
-  @Nonnull def extractSparkArgumentsAsJava(arguments: String): JavaList[String] = {
-    val list = new JavaArrayList[String]()
-    if (StringUtils.isEmpty(arguments)) list
-    else {
-      arguments.split(SPARK_ARGUMENT_REGEXP) match {
-        case d if Utils.isNotEmpty(d) =>
-          d.foreach(x => {
-            if (x.nonEmpty) {
-              list.add(x)
-            }
-          })
-        case _ =>
-      }
-      list
-    }
-  }
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/SparkApplicationActionServiceImpl.java

@@ -80,6 +80,7 @@
 import org.springframework.stereotype.Service;
 
 import java.io.File;
+import java.util.Arrays;
 import java.util.Date;
 import java.util.EnumSet;
 import java.util.HashMap;
@@ -309,6 +310,7 @@ public void start(SparkApplication appParam, boolean auto) throws Exception {
 
         // Get the args after placeholder replacement
         String applicationArgs = variableService.replaceVariable(application.getTeamId(), application.getAppArgs());
+        List<String> sparkArgs = Arrays.asList(applicationArgs.split("\\s+"));
 
         SubmitRequest submitRequest = new SubmitRequest(
             sparkEnv.getSparkVersion(),
@@ -320,7 +322,7 @@ public void start(SparkApplication appParam, boolean auto) throws Exception {
             application.getMainClass(),
             appConf,
             PropertiesUtils.extractSparkPropertiesAsJava(application.getAppProperties()),
-            PropertiesUtils.extractSparkArgumentsAsJava(applicationArgs),
+            sparkArgs,
             application.getApplicationType(),
             application.getHadoopUser(),
             buildResult,

streampark-spark/streampark-spark-client/streampark-spark-client-core/src/main/scala/org/apache/streampark/spark/client/impl/YarnClient.scala

@@ -131,9 +131,8 @@ object YarnClient extends SparkClientTrait {
       .setAppResource(submitRequest.userJarPath)
       .setMainClass(submitRequest.appMain)
       .setAppName(submitRequest.appName)
-      .setConf(
-        "spark.yarn.jars",
-        submitRequest.hdfsWorkspace.sparkLib + "/*.jar")
+      .setConf("spark.yarn.dist.jars", submitRequest.hdfsWorkspace.sparkLib)
+      .setConf("spark.yarn.applicationType", "StreamPark Spark")
       .setVerbose(true)
       .setMaster("yarn")
       .setDeployMode(submitRequest.executionMode match {
@@ -166,7 +165,7 @@ object YarnClient extends SparkClientTrait {
     }
   }
 
-  protected def setYarnQueue(submitRequest: SubmitRequest): Unit = {
+  private def setYarnQueue(submitRequest: SubmitRequest): Unit = {
     if (submitRequest.hasExtra(KEY_SPARK_YARN_QUEUE_NAME)) {
       submitRequest.appProperties.put(KEY_SPARK_YARN_QUEUE, submitRequest.getExtra(KEY_SPARK_YARN_QUEUE_NAME).asInstanceOf[String])
     }