From 0a314c5a8f63acb2059a2e8799643ca5477887eb Mon Sep 17 00:00:00 2001
From: benjobs <benjobx@gmail.com>
Date: Thu, 26 Oct 2023 12:23:17 +0800
Subject: [PATCH 1/7] [Improve] custom-code job support dependencies

---
 .../console/core/entity/Application.java      | 17 ++------
 .../service/impl/AppBuildPipeServiceImpl.java | 41 ++++++++----------
 .../service/impl/ApplicationServiceImpl.java  | 22 +++++++++-
 .../src/main/resources/db/data-h2.sql         |  2 +-
 .../src/main/resources/db/schema-h2.sql       |  1 +
 .../mapper/core/ApplicationMapper.xml         |  1 +
 .../Application/src/AppDarkModeToggle.vue     |  4 +-
 .../ContextMenu/src/ContextMenu.vue           |  4 +-
 .../src/components/Form/src/BasicForm.vue     |  2 +-
 .../src/components/Page/src/PageFooter.vue    |  4 +-
 .../src/hooks/web/useLockPage.ts              |  9 ++--
 .../src/utils/props.ts                        |  2 +-
 .../src/views/flink/app/Add.vue               | 25 +++++------
 .../src/views/flink/app/EditFlink.vue         | 30 ++++++++++++-
 .../src/views/flink/app/EditStreamPark.vue    |  2 +-
 .../flink/app/hooks/useCreateAndEditSchema.ts | 16 +++++--
 .../views/flink/app/hooks/useCreateSchema.ts  | 12 +-----
 .../flink/app/hooks/useEditFlinkSchema.ts     |  8 ++++
 .../views/flink/app/hooks/useFlinkRender.tsx  |  2 +-
 .../flink/packer/maven/MavenTool.scala        | 43 +++++++++++++------
 20 files changed, 149 insertions(+), 98 deletions(-)

diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
index 81f4ccc4cc..3e579c4edd 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
@@ -220,7 +220,7 @@ public class Application implements Serializable {
   /** running job */
   private transient JobsOverview.Task overview;
 
-  private transient String dependency;
+  private String dependency;
   private transient Long sqlId;
   private transient String flinkSql;
 
@@ -390,15 +390,6 @@ public boolean cpFailedTrigger() {
         && this.cpFailureAction != null;
   }
 
-  public boolean eqFlinkJob(Application other) {
-    if (this.isFlinkSqlJob() && other.isFlinkSqlJob()) {
-      if (this.getFlinkSql().trim().equals(other.getFlinkSql().trim())) {
-        return this.getDependencyObject().eq(other.getDependencyObject());
-      }
-    }
-    return false;
-  }
-
   /** Local compilation and packaging working directory */
   @JsonIgnore
   public String getDistHome() {
@@ -752,11 +743,11 @@ public int hashCode() {
 
     @Override
     public String toString() {
-      return groupId + ":" + artifactId + ":" + version + getClassifier(":");
+      return groupId + ":" + artifactId + ":" + version + getClassifier();
     }
 
-    private String getClassifier(String joiner) {
-      return StringUtils.isEmpty(classifier) ? "" : joiner + classifier;
+    private String getClassifier() {
+      return StringUtils.isEmpty(classifier) ? "" : ":" + classifier;
     }
   }
 }
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
index 6f8890bd23..8dc14141dc 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
@@ -184,32 +184,24 @@ public void onStart(PipeSnapshot snapshot) {
 
             if (app.isCustomCodeJob()) {
               // customCode upload jar to appHome...
-              String appHome = app.getAppHome();
               FsOperator fsOperator = app.getFsOperator();
-              fsOperator.delete(appHome);
-              if (app.isUploadJob()) {
+              if (app.isCICDJob()) {
+                String appHome = app.getAppHome();
+                fsOperator.mkCleanDirs(appHome);
+                fsOperator.upload(app.getDistHome(), appHome);
+              } else {
                 File localJar = new File(WebUtils.getAppTempDir(), app.getJar());
                 // upload jar copy to appHome
                 String uploadJar = appUploads.concat("/").concat(app.getJar());
                 checkOrElseUploadJar(app.getFsOperator(), localJar, uploadJar, appUploads);
-                switch (app.getApplicationType()) {
-                  case STREAMPARK_FLINK:
-                    fsOperator.mkdirs(app.getAppLib());
-                    fsOperator.copy(uploadJar, app.getAppLib(), false, true);
-                    break;
-                  case APACHE_FLINK:
-                    fsOperator.mkdirs(appHome);
-                    fsOperator.copy(uploadJar, appHome, false, true);
-                    break;
-                  default:
-                    throw new IllegalArgumentException(
-                        "[StreamPark] unsupported ApplicationType of custom code: "
-                            + app.getApplicationType());
+                if (app.getApplicationType() == ApplicationType.STREAMPARK_FLINK) {
+                  fsOperator.mkdirs(app.getAppLib());
+                  fsOperator.copy(uploadJar, app.getAppLib(), false, true);
                 }
-              } else {
-                fsOperator.upload(app.getDistHome(), appHome);
               }
-            } else {
+            }
+
+            if (app.isFlinkSqlJob() || app.isUploadJob()) {
               if (!app.getDependencyObject().getJar().isEmpty()) {
                 String localUploads = Workspace.local().APP_UPLOADS();
                 // copy jar to local upload dir
@@ -335,7 +327,8 @@ private BuildPipeline createPipelineInstance(@Nonnull Application app) {
     FlinkEnv flinkEnv = flinkEnvService.getByIdOrDefault(app.getVersionId());
     String flinkUserJar = retrieveFlinkUserJar(flinkEnv, app);
     ExecutionMode executionMode = app.getExecutionModeEnum();
-    String mainClass = ConfigConst.STREAMPARK_FLINKSQL_CLIENT_CLASS();
+    String mainClass =
+        app.isCustomCodeJob() ? app.getMainClass() : ConfigConst.STREAMPARK_FLINKSQL_CLIENT_CLASS();
     switch (executionMode) {
       case YARN_APPLICATION:
         String yarnProvidedPath = app.getAppLib();
@@ -358,13 +351,17 @@ private BuildPipeline createPipelineInstance(@Nonnull Application app) {
       case YARN_PER_JOB:
       case YARN_SESSION:
       case REMOTE:
+        boolean skipBuild = app.isCustomCodeJob();
+        if (skipBuild && app.isUploadJob()) {
+          skipBuild = app.getDependencyObject().isEmpty();
+        }
         FlinkRemotePerJobBuildRequest buildRequest =
             new FlinkRemotePerJobBuildRequest(
                 app.getJobName(),
                 app.getLocalAppHome(),
                 mainClass,
                 flinkUserJar,
-                app.isCustomCodeJob(),
+                skipBuild,
                 app.getExecutionModeEnum(),
                 app.getDevelopmentMode(),
                 flinkEnv.getFlinkVersion(),
@@ -424,7 +421,7 @@ private String retrieveFlinkUserJar(FlinkEnv flinkEnv, Application app) {
           case STREAMPARK_FLINK:
             return String.format("%s/%s", app.getAppLib(), app.getModule().concat(".jar"));
           case APACHE_FLINK:
-            return String.format("%s/%s", app.getAppHome(), app.getJar());
+            return String.format("%s/%s", WebUtils.getAppTempDir(), app.getJar());
           default:
             throw new IllegalArgumentException(
                 "[StreamPark] unsupported ApplicationType of custom code: "
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 2242b897e2..4dc07ebaba 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
@@ -703,6 +703,7 @@ else if (ExecutionMode.isKubernetesMode(appParam.getExecutionMode())
   public boolean create(Application appParam) {
     ApiAlertException.throwIfNull(
         appParam.getTeamId(), "The teamId can't be null. Create application failed.");
+    appParam.setBuild(true);
     appParam.setUserId(commonService.getUserId());
     appParam.setState(FlinkAppState.ADDED.getValue());
     appParam.setRelease(ReleaseState.NEED_RELEASE.get());
@@ -736,6 +737,17 @@ public boolean create(Application appParam) {
     }
   }
 
+  @Override
+  public boolean save(Application entity) {
+    String dependency = entity.getDependency();
+    if (entity.isFlinkSqlJob()) {
+      entity.setDependency(null);
+    }
+    boolean flag = super.save(entity);
+    entity.setDependency(dependency);
+    return flag;
+  }
+
   private boolean existsByJobName(String jobName) {
     return this.baseMapper.existsByJobName(jobName);
   }
@@ -837,7 +849,15 @@ public boolean update(Application appParam) {
 
     application.setRelease(ReleaseState.NEED_RELEASE.get());
     if (application.isUploadJob()) {
-      if (!ObjectUtils.safeEquals(application.getJar(), appParam.getJar())) {
+      Application.Dependency thisDependency =
+          Application.Dependency.toDependency(appParam.getDependency());
+      Application.Dependency targetDependency =
+          Application.Dependency.toDependency(application.getDependency());
+
+      boolean depDifference = !thisDependency.eq(targetDependency);
+      if (depDifference) {
+        application.setBuild(true);
+      } else if (!ObjectUtils.safeEquals(application.getJar(), appParam.getJar())) {
         application.setBuild(true);
       } else {
         File jarFile = new File(WebUtils.getAppTempDir(), appParam.getJar());
diff --git a/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql b/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
index 77494579a0..7a450f1292 100644
--- a/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
+++ b/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
@@ -25,7 +25,7 @@ insert into `t_team` values (100001, 'test', 'The test team', now(), now());
 -- ----------------------------
 -- Records of t_flink_app
 -- ----------------------------
-insert into `t_flink_app` values (100000, 100000, 2, 4, null, null, 'Flink SQL Demo', null, null, null, null, null, null , null, 100000, null, 1, null, null, null, null, null, null, null, '0', 0, null, null, null, null, null, null, 'Flink SQL Demo', 0, null, null, null, null, null, null, null, 0, 0, now(), now(), null, 1, 1, null, null, null, null, null, null, 0, null, null, null, 'streampark,test');
+insert into `t_flink_app` values (100000, 100000, 2, 4, null, null, 'Flink SQL Demo', null, null, null, null, null, null, null , null, 100000, null, 1, null, null, null, null, null, null, null, '0', 0, null, null, null, null, null, null, 'Flink SQL Demo', 0, null, null, null, null, null, null, null, 0, 0, now(), now(), null, 1, 1, null, null, null, null, null, null, 0, null, null, null, 'streampark,test');
 
 -- ----------------------------
 -- Records of t_flink_effective
diff --git a/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql b/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql
index 3f7025a83b..d06f907e07 100644
--- a/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql
+++ b/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql
@@ -46,6 +46,7 @@ create table if not exists `t_flink_app` (
   `jar` varchar(255)  default null,
   `jar_check_sum` bigint default null,
   `main_class` varchar(255)  default null,
+  `dependency` text ,
   `args` text,
   `options` text,
   `hot_params` text ,
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml
index a6c97121cb..9096158b55 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml
@@ -40,6 +40,7 @@
         <result column="tracking" jdbcType="INTEGER" property="tracking"/>
         <result column="jar" jdbcType="VARCHAR" property="jar"/>
         <result column="jar_check_sum" jdbcType="VARCHAR" property="jarCheckSum"/>
+        <result column="dependency" jdbcType="LONGVARCHAR" property="dependency"/>
         <result column="main_class" jdbcType="VARCHAR" property="mainClass"/>
         <result column="job_id" jdbcType="VARCHAR" property="jobId"/>
         <result column="job_manager_url" jdbcType="VARCHAR" property="jobManagerUrl"/>
diff --git a/streampark-console/streampark-console-webapp/src/components/Application/src/AppDarkModeToggle.vue b/streampark-console/streampark-console-webapp/src/components/Application/src/AppDarkModeToggle.vue
index abf6c47ea6..40b5c4964b 100644
--- a/streampark-console/streampark-console-webapp/src/components/Application/src/AppDarkModeToggle.vue
+++ b/streampark-console/streampark-console-webapp/src/components/Application/src/AppDarkModeToggle.vue
@@ -79,9 +79,7 @@
       height: 18px;
       background-color: #fff;
       border-radius: 50%;
-      transition:
-        transform 0.5s,
-        background-color 0.5s;
+      transition: transform 0.5s, background-color 0.5s;
       will-change: transform;
     }
 
diff --git a/streampark-console/streampark-console-webapp/src/components/ContextMenu/src/ContextMenu.vue b/streampark-console/streampark-console-webapp/src/components/ContextMenu/src/ContextMenu.vue
index 61cc99b7ff..122b6e711d 100644
--- a/streampark-console/streampark-console-webapp/src/components/ContextMenu/src/ContextMenu.vue
+++ b/streampark-console/streampark-console-webapp/src/components/ContextMenu/src/ContextMenu.vue
@@ -195,9 +195,7 @@
     background-color: @component-background;
     border: 1px solid rgb(0 0 0 / 8%);
     border-radius: 0.25rem;
-    box-shadow:
-      0 2px 2px 0 rgb(0 0 0 / 14%),
-      0 3px 1px -2px rgb(0 0 0 / 10%),
+    box-shadow: 0 2px 2px 0 rgb(0 0 0 / 14%), 0 3px 1px -2px rgb(0 0 0 / 10%),
       0 1px 5px 0 rgb(0 0 0 / 6%);
     background-clip: padding-box;
     user-select: none;
diff --git a/streampark-console/streampark-console-webapp/src/components/Form/src/BasicForm.vue b/streampark-console/streampark-console-webapp/src/components/Form/src/BasicForm.vue
index ab4c106383..0169b4c868 100644
--- a/streampark-console/streampark-console-webapp/src/components/Form/src/BasicForm.vue
+++ b/streampark-console/streampark-console-webapp/src/components/Form/src/BasicForm.vue
@@ -129,7 +129,7 @@
       });
 
       const getBindValue = computed(
-        () => ({ ...attrs, ...props, ...unref(getProps) }) as Recordable,
+        () => ({ ...attrs, ...props, ...unref(getProps) } as Recordable),
       );
 
       const getSchema = computed((): FormSchema[] => {
diff --git a/streampark-console/streampark-console-webapp/src/components/Page/src/PageFooter.vue b/streampark-console/streampark-console-webapp/src/components/Page/src/PageFooter.vue
index c499ccf65c..1533026465 100644
--- a/streampark-console/streampark-console-webapp/src/components/Page/src/PageFooter.vue
+++ b/streampark-console/streampark-console-webapp/src/components/Page/src/PageFooter.vue
@@ -55,9 +55,7 @@
     line-height: 44px;
     background-color: @component-background;
     border-top: 1px solid @border-color-base;
-    box-shadow:
-      0 -6px 16px -8px rgb(0 0 0 / 8%),
-      0 -9px 28px 0 rgb(0 0 0 / 5%),
+    box-shadow: 0 -6px 16px -8px rgb(0 0 0 / 8%), 0 -9px 28px 0 rgb(0 0 0 / 5%),
       0 -12px 48px 16px rgb(0 0 0 / 3%);
     transition: width 0.2s;
 
diff --git a/streampark-console/streampark-console-webapp/src/hooks/web/useLockPage.ts b/streampark-console/streampark-console-webapp/src/hooks/web/useLockPage.ts
index 47aa6eb5f0..de03808115 100644
--- a/streampark-console/streampark-console-webapp/src/hooks/web/useLockPage.ts
+++ b/streampark-console/streampark-console-webapp/src/hooks/web/useLockPage.ts
@@ -48,12 +48,9 @@ export function useLockPage() {
     }
     clear();
 
-    timeId = setTimeout(
-      () => {
-        lockPage();
-      },
-      lockTime * 60 * 1000,
-    );
+    timeId = setTimeout(() => {
+      lockPage();
+    }, lockTime * 60 * 1000);
   }
 
   function lockPage(): void {
diff --git a/streampark-console/streampark-console-webapp/src/utils/props.ts b/streampark-console/streampark-console-webapp/src/utils/props.ts
index 5828c8cf79..92fe2810b1 100644
--- a/streampark-console/streampark-console-webapp/src/utils/props.ts
+++ b/streampark-console/streampark-console-webapp/src/utils/props.ts
@@ -191,7 +191,7 @@ export const buildProps = <
       : never;
   };
 
-export const definePropType = <T>(val: any) => ({ [wrapperKey]: val }) as PropWrapper<T>;
+export const definePropType = <T>(val: any) => ({ [wrapperKey]: val } as PropWrapper<T>);
 
 export const keyOf = <T extends Object>(arr: T) => Object.keys(arr) as Array<keyof T>;
 export const mutable = <T extends readonly any[] | Record<string, unknown>>(val: T) =>
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
index 541503f693..071231e0ca 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
@@ -185,7 +185,7 @@
     // common params...
     const resourceFrom = values.resourceFrom;
     if (resourceFrom) {
-      if (resourceFrom === 'csv') {
+      if (resourceFrom === 'cvs') {
         params['resourceFrom'] = ResourceFromEnum.CICD;
         //streampark flink
         if (values.appType == AppTypeEnum.STREAMPARK_FLINK) {
@@ -210,15 +210,15 @@
           appType: AppTypeEnum.APACHE_FLINK,
           jar: unref(uploadJar),
           mainClass: values.mainClass,
+          dependency: await getDependency(),
         });
         handleCreateApp(params);
       }
     }
   }
-  /* flink sql mode */
-  async function handleSubmitSQL(values: Recordable) {
+  async function getDependency() {
     // Trigger a pom confirmation operation.
-    await unref(dependencyRef)?.handleApplyPom();
+    unref(dependencyRef)?.handleApplyPom();
     // common params...
     const dependency: { pom?: string; jar?: string } = {};
     const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
@@ -233,14 +233,18 @@
         jar: unref(uploadJars),
       });
     }
-
+    return dependency.pom === undefined && dependency.jar === undefined
+      ? null
+      : JSON.stringify(dependency);
+  }
+  /* flink sql mode */
+  async function handleSubmitSQL(values: Recordable) {
     let config = values.configOverride;
-    if (config != null && config !== undefined && config.trim() != '') {
+    if (config != null && config.trim() != '') {
       config = encryptByBase64(config);
     } else {
       config = null;
     }
-
     handleCluster(values);
     const params = {
       jobType: JobTypeEnum.SQL,
@@ -248,10 +252,7 @@
       appType: AppTypeEnum.STREAMPARK_FLINK,
       config,
       format: values.isSetConfig ? 1 : null,
-      dependency:
-        dependency.pom === undefined && dependency.jar === undefined
-          ? null
-          : JSON.stringify(dependency),
+      dependency: await getDependency(),
     };
     handleSubmitParams(params, values, k8sTemplate);
     handleCreateApp(params);
@@ -285,7 +286,7 @@
     const param = {};
     for (const k in params) {
       const v = params[k];
-      if (v != null && v !== undefined) {
+      if (v != null) {
         param[k] = v;
       }
     }
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
index 0213540204..265c66e52f 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
@@ -39,6 +39,7 @@
   import VariableReview from './components/VariableReview.vue';
   import { useDrawer } from '/@/components/Drawer';
   import { ExecModeEnum, ResourceFromEnum } from '/@/enums/flinkEnum';
+  import Dependency from '/@/views/flink/app/components/Dependency.vue';
 
   const route = useRoute();
   const { t } = useI18n();
@@ -52,6 +53,7 @@
   const uploadJar = ref('');
   const programArgRef = ref();
   const podTemplateRef = ref();
+  const dependencyRef = ref();
 
   const k8sTemplate = reactive({
     podTemplate: '',
@@ -116,6 +118,7 @@
       setFieldsValue(defaultParams);
       app.args && programArgRef.value?.setContent(app.args);
       setTimeout(() => {
+        unref(dependencyRef)?.setDefaultValue(JSON.parse(app.dependency || '{}'));
         unref(podTemplateRef)?.handleChoicePodTemplate('ptVisual', app.k8sPodTemplate);
         unref(podTemplateRef)?.handleChoicePodTemplate('jmPtVisual', app.k8sJmPodTemplate);
         unref(podTemplateRef)?.handleChoicePodTemplate('tmPtVisual', app.k8sTmPodTemplate);
@@ -142,15 +145,34 @@
 
   /* Handling update parameters */
   function handleAppUpdate(values: Recordable) {
+    // Trigger a pom confirmation operation.
+    unref(dependencyRef)?.handleApplyPom();
+    // common params...
+    const dependency: { pom?: string; jar?: string } = {};
+    const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
+    const uploadJars = unref(dependencyRef)?.uploadJars;
+    if (unref(dependencyRecords) && unref(dependencyRecords).length > 0) {
+      Object.assign(dependency, {
+        pom: unref(dependencyRecords),
+      });
+    }
+    if (uploadJars && unref(uploadJars).length > 0) {
+      Object.assign(dependency, {
+        jar: unref(uploadJars),
+      });
+    }
     submitLoading.value = true;
     try {
       const params = {
         id: app.id,
         jar: values.jar,
         mainClass: values.mainClass,
+        dependency:
+          dependency.pom === undefined && dependency.jar === undefined
+            ? null
+            : JSON.stringify(dependency),
       };
       handleSubmitParams(params, values, k8sTemplate);
-
       handleUpdateApp(params);
     } catch (error) {
       submitLoading.value = false;
@@ -212,13 +234,17 @@
       <template #args="{ model }">
         <ProgramArgs
           ref="programArgRef"
-          v-if="model.args != null && model.args != undefined"
+          v-if="model.args != null && true"
           v-model:value="model.args"
           :suggestions="suggestions"
           @preview="(value) => openReviewDrawer(true, { value, suggestions })"
         />
       </template>
 
+      <template #dependency="{ model, field }">
+        <Dependency ref="dependencyRef" v-model:value="model[field]" :form-model="model" />
+      </template>
+
       <template #formFooter>
         <div class="flex items-center w-full justify-center">
           <a-button @click="go('/flink/app')">
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
index 0d346aa10e..9e307ea244 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
@@ -360,7 +360,7 @@
       <template #args="{ model }">
         <ProgramArgs
           ref="programArgRef"
-          v-if="model.args != null && model.args != undefined"
+          v-if="model.args != null"
           v-model:value="model.args"
           :suggestions="suggestions"
           @preview="(value) => openReviewDrawer(true, { value, suggestions })"
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
index a5cad757ce..752ad30e0e 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
@@ -49,7 +49,13 @@ import { fetchFlinkEnv } from '/@/api/flink/setting/flinkEnv';
 import { FlinkEnv } from '/@/api/flink/setting/types/flinkEnv.type';
 import { AlertSetting } from '/@/api/flink/setting/types/alert.type';
 import { FlinkCluster } from '/@/api/flink/setting/types/flinkCluster.type';
-import { AppTypeEnum, ClusterStateEnum, ExecModeEnum, JobTypeEnum } from '/@/enums/flinkEnum';
+import {
+  AppTypeEnum,
+  ClusterStateEnum,
+  ExecModeEnum,
+  JobTypeEnum,
+  ResourceFromEnum,
+} from '/@/enums/flinkEnum';
 import { isK8sExecMode } from '../utils';
 import { useI18n } from '/@/hooks/web/useI18n';
 import { fetchCheckHadoop } from '/@/api/flink/setting';
@@ -79,7 +85,7 @@ export const useCreateAndEditSchema = (
 
   const [registerConfDrawer, { openDrawer: openConfDrawer }] = useDrawer();
 
-  /* 
+  /*
   !The original item is also unassigned
   */
   function getConfigSchemas() {
@@ -126,9 +132,11 @@ export const useCreateAndEditSchema = (
         slot: 'dependency',
         ifShow: ({ values }) => {
           if (edit?.appId) {
-            return values.jobType == JobTypeEnum.SQL;
+            return values.jobType == JobTypeEnum.SQL
+              ? true
+              : values.resourceFrom == ResourceFromEnum.UPLOAD;
           } else {
-            return values?.jobType == 'sql';
+            return values?.jobType == 'sql' ? true : values?.resourceFrom != 'cvs';
           }
         },
       },
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts
index 72b11762f1..4ef2557b35 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts
@@ -63,14 +63,6 @@ export const useCreateSchema = (dependencyRef: Ref) => {
     suggestions,
   } = useCreateAndEditSchema(dependencyRef);
 
-  // async function handleEditConfig(config: string) {
-  //   console.log('config', config);
-  //   const res = await fetchAppConf({ config });
-  //   const conf = decodeByBase64(res);
-  //   openConfDrawer(true, {
-  //     configOverride: conf,
-  //   });
-  // }
   function handleCheckConfig(_rule: RuleObject, value: StoreValue) {
     if (value) {
       const confType = getAppConfType(value);
@@ -97,7 +89,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
               if (value === 'sql') {
                 formModel.tableEnv = 1;
               } else {
-                formModel.resourceFrom = 'csv';
+                formModel.resourceFrom = 'cvs';
               }
             },
           };
@@ -109,7 +101,6 @@ export const useCreateSchema = (dependencyRef: Ref) => {
       },
       ...getExecutionModeSchema.value,
       ...getFlinkClusterSchemas.value,
-      ...getFlinkSqlSchema.value,
       {
         field: 'resourceFrom',
         label: t('flink.app.resourceFrom'),
@@ -133,6 +124,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
         ifShow: ({ values }) => values?.jobType !== 'sql' && values?.resourceFrom == 'upload',
         rules: [{ required: true, message: t('flink.app.addAppTips.mainClassIsRequiredMessage') }],
       },
+      ...getFlinkSqlSchema.value,
       {
         field: 'project',
         label: t('flink.app.project'),
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useEditFlinkSchema.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useEditFlinkSchema.ts
index aef6f6f81f..c7e072ad83 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useEditFlinkSchema.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useEditFlinkSchema.ts
@@ -23,6 +23,8 @@ import { Alert } from 'ant-design-vue';
 import { useRoute } from 'vue-router';
 import { fetchMain } from '/@/api/flink/app/app';
 import { ResourceFromEnum } from '/@/enums/flinkEnum';
+import { useI18n } from '/@/hooks/web/useI18n';
+const { t } = useI18n();
 
 export const useEditFlinkSchema = (jars: Ref) => {
   const flinkSql = ref();
@@ -115,6 +117,12 @@ export const useEditFlinkSchema = (jars: Ref) => {
         },
         rules: [{ required: true, message: 'Program Main is required' }],
       },
+      {
+        field: 'dependency',
+        label: t('flink.app.dependency'),
+        component: 'Input',
+        slot: 'dependency',
+      },
       ...getFlinkFormOtherSchemas.value,
     ];
   });
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx
index 4a103f81be..ba00e667a8 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx
@@ -514,7 +514,7 @@ export const renderResourceFrom = (model: Recordable) => {
       value={model.resourceFrom}
       placeholder="Please select resource from"
     >
-      <Select.Option value="csv">
+      <Select.Option value="cvs">
         <SvgIcon name="github" />
         <span class="pl-10px">CICD</span>
         <span class="gray">(build from CVS)</span>
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/maven/MavenTool.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/maven/MavenTool.scala
index 9c95748dd2..5dab91892a 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/maven/MavenTool.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/maven/MavenTool.scala
@@ -46,7 +46,7 @@ import java.util
 
 import scala.collection.JavaConversions._
 import scala.collection.mutable.ArrayBuffer
-import scala.util.Try
+import scala.util.{Failure, Success, Try}
 
 object MavenTool extends Logger {
 
@@ -76,9 +76,6 @@ object MavenTool extends Logger {
     List(remoteRepository)
   }
 
-  private val isJarFile = (file: File) =>
-    file.isFile && Try(Utils.checkJarFile(file.toURI.toURL)).isSuccess
-
   /**
    * Build a fat-jar with custom jar libraries.
    *
@@ -102,16 +99,34 @@ object MavenTool extends Logger {
     uberJar.delete()
     // resolve all jarLibs
     val jarSet = new util.HashSet[File]
-    jarLibs
-      .map(lib => new File(lib))
-      .filter(_.exists)
-      .foreach {
-        case libFile if isJarFile(libFile) => jarSet.add(libFile)
-        case libFile if libFile.isDirectory =>
-          libFile.listFiles.filter(isJarFile).foreach(jarSet.add)
-        case _ =>
-      }
-    logInfo(s"start shaded fat-jar: ${jarLibs.mkString(",")}")
+    jarLibs.foreach {
+      x =>
+        new File(x) match {
+          case jarFile if jarFile.exists() =>
+            if (jarFile.isFile) {
+              Try(Utils.checkJarFile(jarFile.toURI.toURL)) match {
+                case Success(_) => jarSet.add(jarFile)
+                case Failure(e) => logWarn(s"buildFatJar: error, ${e.getMessage}")
+              }
+            } else {
+              jarFile.listFiles.foreach(
+                jar => {
+                  if (jar.isFile) {
+                    Try(Utils.checkJarFile(jar.toURI.toURL)) match {
+                      case Success(_) => jarSet.add(jar)
+                      case Failure(e) =>
+                        logWarn(
+                          s"buildFatJar: directory [${jarFile.getAbsolutePath}], error: ${e.getMessage}")
+                    }
+                  }
+                })
+            }
+          case _ =>
+        }
+    }
+
+    logInfo(s"start shaded fat-jar: ${jarSet.mkString(",")}")
+
     // shade jars
     val shadeRequest = {
       val req = new ShadeRequest

From 68d7865f18c9caa8c4b4ec40103470e9c7ebe4fb Mon Sep 17 00:00:00 2001
From: benjobs <benjobx@gmail.com>
Date: Thu, 26 Oct 2023 12:34:30 +0800
Subject: [PATCH 2/7] [Improve] ddl improvement

---
 .../assembly/script/upgrade/mysql/2.1.1.sql   | 26 +++++++++++++++++++
 .../assembly/script/upgrade/pgsql/2.1.1.sql   | 19 ++++++++++++++
 .../console/core/entity/Application.java      |  4 ++-
 .../src/views/flink/app/EditFlink.vue         |  2 +-
 4 files changed, 49 insertions(+), 2 deletions(-)
 create mode 100644 streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.1.sql
 create mode 100644 streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.1.sql

diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.1.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.1.sql
new file mode 100644
index 0000000000..2205a8fbb6
--- /dev/null
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.1.sql
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+
+use streampark;
+
+set names utf8mb4;
+set foreign_key_checks = 0;
+
+alter table `t_flink_app`
+    add column `dependency` text collate utf8mb4_general_ci default null after `main_class`;
+
+set foreign_key_checks = 1;
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.1.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.1.sql
new file mode 100644
index 0000000000..a87816bf48
--- /dev/null
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.1.sql
@@ -0,0 +1,19 @@
+/*
+* 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.
+*/
+
+alter table "public"."t_flink_app"
+    add column "dependency" text collate "pg_catalog"."default";
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
index 3e579c4edd..568f7f2a52 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
@@ -160,6 +160,9 @@ public class Application implements Serializable {
 
   private String mainClass;
 
+  @TableField(updateStrategy = FieldStrategy.IGNORED)
+  private String dependency;
+
   private Date startTime;
 
   @TableField(updateStrategy = FieldStrategy.IGNORED)
@@ -220,7 +223,6 @@ public class Application implements Serializable {
   /** running job */
   private transient JobsOverview.Task overview;
 
-  private String dependency;
   private transient Long sqlId;
   private transient String flinkSql;
 
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
index 265c66e52f..e75e0d544f 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
@@ -234,7 +234,7 @@
       <template #args="{ model }">
         <ProgramArgs
           ref="programArgRef"
-          v-if="model.args != null && true"
+          v-if="model.args != null"
           v-model:value="model.args"
           :suggestions="suggestions"
           @preview="(value) => openReviewDrawer(true, { value, suggestions })"

From 596e300f0c0d038fda42738aa2966f0fc306efd4 Mon Sep 17 00:00:00 2001
From: benjobs <benjobx@gmail.com>
Date: Thu, 26 Oct 2023 12:46:19 +0800
Subject: [PATCH 3/7] minor improve

---
 .../console/core/service/impl/ApplicationServiceImpl.java       | 2 ++
 1 file changed, 2 insertions(+)

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 4dc07ebaba..d3f50a6e0a 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
@@ -849,6 +849,8 @@ public boolean update(Application appParam) {
 
     application.setRelease(ReleaseState.NEED_RELEASE.get());
     if (application.isUploadJob()) {
+      application.setDependency(appParam.getDependency());
+
       Application.Dependency thisDependency =
           Application.Dependency.toDependency(appParam.getDependency());
       Application.Dependency targetDependency =

From c9affadc3a65484c2b654598e7ab5ed7a3fa539e Mon Sep 17 00:00:00 2001
From: benjobs <benjobx@gmail.com>
Date: Thu, 26 Oct 2023 14:02:32 +0800
Subject: [PATCH 4/7] [Improve] build fatjar improvement

---
 .../service/impl/AppBuildPipeServiceImpl.java |  5 ---
 .../service/impl/ApplicationServiceImpl.java  | 19 +++++++---
 .../flink/packer/pipeline/BuildRequest.scala  |  1 -
 .../impl/FlinkRemoteBuildPipeline.scala       | 35 ++++++++-----------
 4 files changed, 28 insertions(+), 32 deletions(-)

diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
index 8dc14141dc..f6fb7e641f 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
@@ -351,17 +351,12 @@ private BuildPipeline createPipelineInstance(@Nonnull Application app) {
       case YARN_PER_JOB:
       case YARN_SESSION:
       case REMOTE:
-        boolean skipBuild = app.isCustomCodeJob();
-        if (skipBuild && app.isUploadJob()) {
-          skipBuild = app.getDependencyObject().isEmpty();
-        }
         FlinkRemotePerJobBuildRequest buildRequest =
             new FlinkRemotePerJobBuildRequest(
                 app.getJobName(),
                 app.getLocalAppHome(),
                 mainClass,
                 flinkUserJar,
-                skipBuild,
                 app.getExecutionModeEnum(),
                 app.getDevelopmentMode(),
                 flinkEnv.getFlinkVersion(),
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 d3f50a6e0a..e23c125647 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
@@ -703,7 +703,17 @@ else if (ExecutionMode.isKubernetesMode(appParam.getExecutionMode())
   public boolean create(Application appParam) {
     ApiAlertException.throwIfNull(
         appParam.getTeamId(), "The teamId can't be null. Create application failed.");
-    appParam.setBuild(true);
+
+    if (appParam.isFlinkSqlJob()) {
+      appParam.setBuild(true);
+    } else {
+      if (appParam.isUploadJob()) {
+        appParam.setBuild(!appParam.getDependencyObject().isEmpty());
+      } else {
+        appParam.setBuild(false);
+      }
+    }
+
     appParam.setUserId(commonService.getUserId());
     appParam.setState(FlinkAppState.ADDED.getValue());
     appParam.setRelease(ReleaseState.NEED_RELEASE.get());
@@ -848,16 +858,15 @@ public boolean update(Application appParam) {
         String.format(ERROR_APP_QUEUE_HINT, appParam.getYarnQueue(), appParam.getTeamId()));
 
     application.setRelease(ReleaseState.NEED_RELEASE.get());
-    if (application.isUploadJob()) {
-      application.setDependency(appParam.getDependency());
 
+    if (application.isUploadJob()) {
       Application.Dependency thisDependency =
           Application.Dependency.toDependency(appParam.getDependency());
       Application.Dependency targetDependency =
           Application.Dependency.toDependency(application.getDependency());
 
-      boolean depDifference = !thisDependency.eq(targetDependency);
-      if (depDifference) {
+      if (!thisDependency.eq(targetDependency)) {
+        application.setDependency(appParam.getDependency());
         application.setBuild(true);
       } else if (!ObjectUtils.safeEquals(application.getJar(), appParam.getJar())) {
         application.setBuild(true);
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
index 41e6e4f461..62f90935a2 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
@@ -107,7 +107,6 @@ case class FlinkRemotePerJobBuildRequest(
     workspace: String,
     mainClass: String,
     customFlinkUserJar: String,
-    skipBuild: Boolean,
     executionMode: ExecutionMode,
     developmentMode: DevelopmentMode,
     flinkVersion: FlinkVersion,
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala
index d6889caa2d..86c523fdd9 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala
@@ -31,28 +31,21 @@ class FlinkRemoteBuildPipeline(request: FlinkRemotePerJobBuildRequest) extends B
   /** The construction logic needs to be implemented by subclasses */
   @throws[Throwable]
   override protected def buildProcess(): ShadedBuildResponse = {
-    // create workspace.
-    // the sub workspace path like: APP_WORKSPACE/jobName
-    if (request.skipBuild) {
-      ShadedBuildResponse(request.workspace, request.customFlinkUserJar)
-    } else {
-      execStep(1) {
-        LfsOperator.mkCleanDirs(request.workspace)
-        logInfo(s"recreate building workspace: ${request.workspace}")
+    execStep(1) {
+      LfsOperator.mkCleanDirs(request.workspace)
+      logInfo(s"recreate building workspace: ${request.workspace}")
+    }.getOrElse(throw getError.exception)
+    // build flink job shaded jar
+    val shadedJar =
+      execStep(2) {
+        val output = MavenTool.buildFatJar(
+          request.mainClass,
+          request.providedLibs,
+          request.getShadedJarPath(request.workspace))
+        logInfo(s"output shaded flink job jar: ${output.getAbsolutePath}")
+        output
       }.getOrElse(throw getError.exception)
-      // build flink job shaded jar
-      val shadedJar =
-        execStep(2) {
-          val output = MavenTool.buildFatJar(
-            request.mainClass,
-            request.providedLibs,
-            request.getShadedJarPath(request.workspace))
-          logInfo(s"output shaded flink job jar: ${output.getAbsolutePath}")
-          output
-        }.getOrElse(throw getError.exception)
-      ShadedBuildResponse(request.workspace, shadedJar.getAbsolutePath)
-    }
-
+    ShadedBuildResponse(request.workspace, shadedJar.getAbsolutePath)
   }
 
 }

From b6faaa1a6db556142338db4741335baddbf35b9c Mon Sep 17 00:00:00 2001
From: benjobs <benjobx@gmail.com>
Date: Thu, 26 Oct 2023 16:45:23 +0800
Subject: [PATCH 5/7] [improve] extractArguments improvement

---
 .../common/util/PropertiesUtils.scala         | 43 ++++++++++-
 .../common/util/PropertiesUtilsTestCase.scala | 66 ++---------------
 .../flink/client/trait/FlinkClientTrait.scala | 71 +++----------------
 3 files changed, 56 insertions(+), 124 deletions(-)

diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/util/PropertiesUtils.scala b/streampark-common/src/main/scala/org/apache/streampark/common/util/PropertiesUtils.scala
index 5443148015..2272cc5cfe 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/util/PropertiesUtils.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/util/PropertiesUtils.scala
@@ -30,7 +30,7 @@ import java.util.regex.Pattern
 import scala.collection.JavaConversions._
 import scala.collection.JavaConverters._
 import scala.collection.mutable
-import scala.collection.mutable.{Map => MutableMap}
+import scala.collection.mutable.{ArrayBuffer, Map => MutableMap}
 
 object PropertiesUtils extends Logger {
 
@@ -305,6 +305,47 @@ object PropertiesUtils extends Logger {
     }
   }
 
+  @Nonnull def extractArguments(args: String): List[String] = {
+    val programArgs = new ArrayBuffer[String]()
+    if (StringUtils.isNotEmpty(args)) {
+      val array = args.split("\\s+")
+      val iter = array.iterator
+      def join(s: String, v: String): Unit = {
+        if (v.startsWith(s)) {
+          if (v.endsWith(s)) {
+            programArgs += v.replaceAll(s"^$s|$s$$", "")
+          } else {
+            var value = v
+            while (!value.endsWith(s) && iter.hasNext) {
+              value += s" ${iter.next()}"
+            }
+            programArgs += value.replaceAll(s"^$s|$s$$", "")
+          }
+        }
+      }
+
+      while (iter.hasNext) {
+        val v = iter.next()
+        if (v.startsWith("'")) {
+          if (v.endsWith("'")) {
+            programArgs += v.replaceAll("^'|'$", "")
+          } else {
+            join("'", v)
+          }
+        } else if (v.startsWith("\"")) {
+          if (v.endsWith("\"")) {
+            programArgs += v.replaceAll("^\"|\"$", "")
+          } else {
+            join("\"", v)
+          }
+        } else {
+          programArgs += v
+        }
+      }
+    }
+    programArgs.toList
+  }
+
   @Nonnull def extractDynamicPropertiesAsJava(properties: String): JavaMap[String, String] =
     new JavaMap[String, String](extractDynamicProperties(properties).asJava)
 
diff --git a/streampark-common/src/test/scala/org/apache/streampark/common/util/PropertiesUtilsTestCase.scala b/streampark-common/src/test/scala/org/apache/streampark/common/util/PropertiesUtilsTestCase.scala
index e04e0e3be3..995cfd2ece 100644
--- a/streampark-common/src/test/scala/org/apache/streampark/common/util/PropertiesUtilsTestCase.scala
+++ b/streampark-common/src/test/scala/org/apache/streampark/common/util/PropertiesUtilsTestCase.scala
@@ -16,75 +16,19 @@
  */
 package org.apache.streampark.common.util
 
-import org.apache.commons.lang3.StringUtils
 import org.junit.jupiter.api.{Assertions, Test}
 
 import scala.collection.mutable.ArrayBuffer
+import scala.language.postfixOps
 
 class PropertiesUtilsTestCase {
 
   @Test def testExtractProgramArgs(): Unit = {
-    val argsStr = "--host localhost:8123\n\n\n" +
-      "--sql \"\"\"insert into table_a select * from table_b\"\"\"\n" +
-      "--c d\r\n" +
-      "--x yyy"
+    val args =
+      "mysql-sync-database \n--database employees \n--mysql-conf hostname=127.0.0.1 \n--mysql-conf port=3306 \n--mysql-conf username=root \n--mysql-conf password=123456 \n--mysql-conf database-name=employees \n--including-tables 'test|test.*' \n--sink-conf fenodes=127.0.0.1:8030 \n--sink-conf username=root \n--sink-conf password= \n--sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \n--sink-conf sink.label-prefix=label\n--table-conf replication_num=1 "
     val programArgs = new ArrayBuffer[String]()
-    if (StringUtils.isNotEmpty(argsStr)) {
-      val multiLineChar = "\"\"\""
-      val array = argsStr.split("\\s+")
-      if (array.filter(_.startsWith(multiLineChar)).isEmpty) {
-        array.foreach(programArgs +=)
-      } else {
-        val argsArray = new ArrayBuffer[String]()
-        val tempBuffer = new ArrayBuffer[String]()
-
-        def processElement(index: Int, multiLine: Boolean): Unit = {
-
-          if (index == array.length) {
-            if (tempBuffer.nonEmpty) {
-              argsArray += tempBuffer.mkString(" ")
-            }
-            return
-          }
-
-          val next = index + 1
-          val elem = array(index)
-
-          if (elem.trim.nonEmpty) {
-            if (!multiLine) {
-              if (elem.startsWith(multiLineChar)) {
-                tempBuffer += elem.drop(3)
-                processElement(next, true)
-              } else {
-                argsArray += elem
-                processElement(next, false)
-              }
-            } else {
-              if (elem.endsWith(multiLineChar)) {
-                tempBuffer += elem.dropRight(3)
-                argsArray += tempBuffer.mkString(" ")
-                tempBuffer.clear()
-                processElement(next, false)
-              } else {
-                tempBuffer += elem
-                processElement(next, multiLine)
-              }
-            }
-          } else {
-            tempBuffer += elem
-            processElement(next, false)
-          }
-        }
-
-        processElement(0, false)
-        argsArray.foreach(x => programArgs += x.trim)
-      }
-    }
-
-    Assertions.assertEquals("localhost:8123", programArgs(1))
-    Assertions.assertEquals("insert into table_a select * from table_b", programArgs(3))
-    Assertions.assertEquals("d", programArgs(5))
-    Assertions.assertEquals("yyy", programArgs(7))
+    programArgs ++= PropertiesUtils.extractArguments(args)
+    println(programArgs)
   }
 
   @Test def testDynamicProperties(): Unit = {
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 aa91aed32b..bed7ed552a 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
@@ -20,7 +20,7 @@ package org.apache.streampark.flink.client.`trait`
 import org.apache.streampark.common.conf.ConfigConst._
 import org.apache.streampark.common.conf.Workspace
 import org.apache.streampark.common.enums.{ApplicationType, DevelopmentMode, ExecutionMode}
-import org.apache.streampark.common.util.{DeflaterUtils, Logger}
+import org.apache.streampark.common.util.{DeflaterUtils, Logger, PropertiesUtils}
 import org.apache.streampark.flink.client.bean._
 import org.apache.streampark.flink.core.FlinkClusterClient
 import org.apache.streampark.flink.core.conf.FlinkRunOption
@@ -40,6 +40,7 @@ import org.apache.flink.util.FlinkException
 import org.apache.flink.util.Preconditions.checkNotNull
 
 import java.io.File
+import java.util
 import java.util.{Collections, List => JavaList, Map => JavaMap}
 
 import scala.annotation.tailrec
@@ -399,75 +400,21 @@ trait FlinkClientTrait extends Logger {
 
   private[this] def extractProgramArgs(submitRequest: SubmitRequest): JavaList[String] = {
     val programArgs = new ArrayBuffer[String]()
-    val args = submitRequest.args
-
-    if (StringUtils.isNotEmpty(args)) {
-      val multiChar = "\""
-      val array = args.split("\\s+")
-      if (!array.exists(_.startsWith(multiChar))) {
-        array.foreach(programArgs +=)
-      } else {
-        val argsArray = new ArrayBuffer[String]()
-        val tempBuffer = new ArrayBuffer[String]()
-
-        @tailrec
-        def processElement(index: Int, multi: Boolean): Unit = {
-
-          if (index == array.length) {
-            if (tempBuffer.nonEmpty) {
-              argsArray += tempBuffer.mkString(" ")
-            }
-            return
-          }
-
-          val next = index + 1
-          val elem = array(index).trim
-
-          if (elem.isEmpty) {
-            processElement(next, multi = false)
-          } else {
-            if (multi) {
-              if (elem.endsWith(multiChar)) {
-                tempBuffer += elem.dropRight(1)
-                argsArray += tempBuffer.mkString(" ")
-                tempBuffer.clear()
-                processElement(next, multi = false)
-              } else {
-                tempBuffer += elem
-                processElement(next, multi)
-              }
-            } else {
-              val until = if (elem.endsWith(multiChar)) 1 else 0
-              if (elem.startsWith(multiChar)) {
-                tempBuffer += elem.drop(1).dropRight(until)
-                processElement(next, multi = true)
-              } else {
-                argsArray += elem.dropRight(until)
-                processElement(next, multi = false)
-              }
-            }
-          }
-        }
-
-        processElement(0, multi = false)
-        argsArray.foreach(x => programArgs += x)
-      }
-    }
+    programArgs ++= PropertiesUtils.extractArguments(submitRequest.args)
 
     if (submitRequest.applicationType == ApplicationType.STREAMPARK_FLINK) {
-
-      programArgs += PARAM_KEY_FLINK_CONF += submitRequest.flinkYaml
-      programArgs += PARAM_KEY_APP_NAME += DeflaterUtils.zipString(submitRequest.effectiveAppName)
-      programArgs += PARAM_KEY_FLINK_PARALLELISM += getParallelism(submitRequest).toString
+      programArgs ++ PARAM_KEY_FLINK_CONF += submitRequest.flinkYaml
+      programArgs ++ PARAM_KEY_APP_NAME += DeflaterUtils.zipString(submitRequest.effectiveAppName)
+      programArgs ++ PARAM_KEY_FLINK_PARALLELISM += getParallelism(submitRequest).toString
 
       submitRequest.developmentMode match {
         case DevelopmentMode.FLINK_SQL =>
-          programArgs += PARAM_KEY_FLINK_SQL += submitRequest.flinkSQL
+          programArgs ++ PARAM_KEY_FLINK_SQL += submitRequest.flinkSQL
           if (submitRequest.appConf != null) {
-            programArgs += PARAM_KEY_APP_CONF += submitRequest.appConf
+            programArgs ++ PARAM_KEY_APP_CONF += submitRequest.appConf
           }
         case _ if Try(!submitRequest.appConf.startsWith("json:")).getOrElse(true) =>
-          programArgs += PARAM_KEY_APP_CONF += submitRequest.appConf
+          programArgs ++ PARAM_KEY_APP_CONF += submitRequest.appConf
       }
 
     }

From a0fa17a4f80870fdffcb2fbac16e671681497aa2 Mon Sep 17 00:00:00 2001
From: benjobs <benjobx@gmail.com>
Date: Thu, 26 Oct 2023 16:48:41 +0800
Subject: [PATCH 6/7] minor improvement

---
 .../flink/client/trait/FlinkClientTrait.scala        | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

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 bed7ed552a..bfcf682dcf 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
@@ -403,18 +403,18 @@ trait FlinkClientTrait extends Logger {
     programArgs ++= PropertiesUtils.extractArguments(submitRequest.args)
 
     if (submitRequest.applicationType == ApplicationType.STREAMPARK_FLINK) {
-      programArgs ++ PARAM_KEY_FLINK_CONF += submitRequest.flinkYaml
-      programArgs ++ PARAM_KEY_APP_NAME += DeflaterUtils.zipString(submitRequest.effectiveAppName)
-      programArgs ++ PARAM_KEY_FLINK_PARALLELISM += getParallelism(submitRequest).toString
+      programArgs += PARAM_KEY_FLINK_CONF += submitRequest.flinkYaml
+      programArgs += PARAM_KEY_APP_NAME += DeflaterUtils.zipString(submitRequest.effectiveAppName)
+      programArgs += PARAM_KEY_FLINK_PARALLELISM += getParallelism(submitRequest).toString
 
       submitRequest.developmentMode match {
         case DevelopmentMode.FLINK_SQL =>
-          programArgs ++ PARAM_KEY_FLINK_SQL += submitRequest.flinkSQL
+          programArgs += PARAM_KEY_FLINK_SQL += submitRequest.flinkSQL
           if (submitRequest.appConf != null) {
-            programArgs ++ PARAM_KEY_APP_CONF += submitRequest.appConf
+            programArgs += PARAM_KEY_APP_CONF += submitRequest.appConf
           }
         case _ if Try(!submitRequest.appConf.startsWith("json:")).getOrElse(true) =>
-          programArgs ++ PARAM_KEY_APP_CONF += submitRequest.appConf
+          programArgs += PARAM_KEY_APP_CONF += submitRequest.appConf
       }
 
     }

From 0ceea0f04bbbce0a85fe3494615cdb05e2c3e2ed Mon Sep 17 00:00:00 2001
From: benjobs <benjobx@gmail.com>
Date: Thu, 26 Oct 2023 18:31:16 +0800
Subject: [PATCH 7/7] minor improve

---
 .../apache/streampark/flink/client/trait/FlinkClientTrait.scala | 2 --
 1 file changed, 2 deletions(-)

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 bfcf682dcf..e7da5525e2 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
@@ -40,10 +40,8 @@ import org.apache.flink.util.FlinkException
 import org.apache.flink.util.Preconditions.checkNotNull
 
 import java.io.File
-import java.util
 import java.util.{Collections, List => JavaList, Map => JavaMap}
 
-import scala.annotation.tailrec
 import scala.collection.JavaConversions._
 import scala.collection.JavaConverters._
 import scala.collection.mutable