[Improve] FE job dependency component improvements (#3373)

* [Improve] FE job dependency component improvements

* [Improve] MavenTool support mirror|proxy

* [Improve] FE dependency component get latest value bug fixed

* [Improve] resolve maven artifacts bug fixed.

---------

Co-authored-by: benjobs <[email protected]>

Author: wolfboys

.mvn/wrapper/MavenWrapperHelper.java

@@ -98,7 +98,7 @@ protected PasswordAuthentication getPasswordAuthentication() {
         log(" - Downloader complete");
     }
 
-    public static String getFileMd5(String path) throws Exception {
+    private static String getFileMd5(String path) throws Exception {
         MessageDigest md5 = MessageDigest.getInstance("MD5");
         try (FileInputStream inputStream = new FileInputStream(path)) {
             byte[] buffer = new byte[1024];

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java

@@ -462,6 +462,12 @@ public boolean isCustomCodeJob() {
     return DevelopmentMode.CUSTOM_CODE.getValue().equals(this.getJobType());
   }
 
+  @JsonIgnore
+  public boolean isApacheFlinkCustomCodeJob() {
+    return DevelopmentMode.CUSTOM_CODE.getValue().equals(this.getJobType())
+        && getApplicationType() == ApplicationType.APACHE_FLINK;
+  }
+
   @JsonIgnore
   public boolean isUploadJob() {
     return isCustomCodeJob() && ResourceFrom.UPLOAD.getValue().equals(this.getResourceFrom());

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

@@ -378,7 +378,7 @@ private void prepareJars(Application app) throws IOException {
 
     FsOperator localFS = FsOperator.lfs();
     // 1. copy jar to local upload dir
-    if (app.isFlinkSqlJob() || app.isUploadJob()) {
+    if (app.isFlinkSqlJob() || app.isApacheFlinkCustomCodeJob()) {
       if (!app.getMavenDependency().getJar().isEmpty()) {
         for (String jar : app.getMavenDependency().getJar()) {
           File localJar = new File(WebUtils.getAppTempDir(), jar);
@@ -393,7 +393,7 @@ private void prepareJars(Application app) throws IOException {
       }
     }
 
-    if (app.isCustomCodeJob()) {
+    if (app.isApacheFlinkCustomCodeJob()) {
       // customCode upload jar to appHome...
       FsOperator fsOperator = app.getFsOperator();
       ResourceFrom resourceFrom = ResourceFrom.of(app.getResourceFrom());
@@ -437,7 +437,15 @@ private void prepareJars(Application app) throws IOException {
         if (!app.getMavenDependency().getPom().isEmpty()) {
           Set<Artifact> artifacts =
               app.getMavenDependency().getPom().stream()
-                  .filter(x -> !new File(localUploadDIR, x.artifactName()).exists())
+                  .filter(
+                      dep -> {
+                        File file = new File(localUploadDIR, dep.artifactName());
+                        if (file.exists()) {
+                          dependencyJars.add(file);
+                          return false;
+                        }
+                        return true;
+                      })
                   .map(
                       pom ->
                           new Artifact(

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

@@ -852,7 +852,7 @@ public boolean update(Application appParam) {
 
     application.setRelease(ReleaseState.NEED_RELEASE.get());
 
-    if (application.isUploadJob()) {
+    if (application.isApacheFlinkCustomCodeJob()) {
       MavenDependency thisDependency = MavenDependency.of(appParam.getDependency());
       MavenDependency targetDependency = MavenDependency.of(application.getDependency());
 

streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue

@@ -201,8 +201,9 @@
         } else {
           params['jar'] = values.jar || null;
           params['mainClass'] = values.mainClass || null;
+          params['dependency'] = await getDependency();
         }
-        handleCreateApp(params);
+        await handleCreateApp(params);
       } else {
         // from upload
         Object.assign(params, {
@@ -212,13 +213,13 @@
           mainClass: values.mainClass,
           dependency: await getDependency(),
         });
-        handleCreateApp(params);
+        await handleCreateApp(params);
       }
     }
   }
   async function getDependency() {
     // Trigger a pom confirmation operation.
-    unref(dependencyRef)?.handleApplyPom();
+    await unref(dependencyRef)?.handleApplyPom();
     // common params...
     const dependency: { pom?: string; jar?: string } = {};
     const dependencyRecords = unref(dependencyRef)?.dependencyRecords;

streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue

@@ -144,9 +144,9 @@
   }
 
   /* Handling update parameters */
-  function handleAppUpdate(values: Recordable) {
+  async function handleAppUpdate(values: Recordable) {
     // Trigger a pom confirmation operation.
-    unref(dependencyRef)?.handleApplyPom();
+    await unref(dependencyRef)?.handleApplyPom();
     // common params...
     const dependency: { pom?: string; jar?: string } = {};
     const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
@@ -173,7 +173,7 @@
             : JSON.stringify(dependency),
       };
       handleSubmitParams(params, values, k8sTemplate);
-      handleUpdateApp(params);
+      await handleUpdateApp(params);
     } catch (error) {
       submitLoading.value = false;
     }

streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue

@@ -176,10 +176,10 @@
     }
   }
 
-  function handleSubmitSQL(values: Recordable) {
+  async function handleSubmitSQL(values: Recordable) {
     try {
       // Trigger a pom confirmation operation.
-      unref(dependencyRef)?.handleApplyPom();
+      await unref(dependencyRef)?.handleApplyPom();
       // common params...
       const dependency: { pom?: string; jar?: string } = {};
       const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
@@ -212,7 +212,7 @@
             : JSON.stringify(dependency),
       };
       handleSubmitParams(params, values, k8sTemplate);
-      handleUpdateApp(params);
+      await handleUpdateApp(params);
     } catch (error) {
       createMessage.error('edit error');
       submitLoading.value = false;

streampark-console/streampark-console-webapp/src/views/flink/app/components/Dependency.vue

@@ -75,7 +75,7 @@
   const { t } = useI18n();
   const defaultValue = '';
   const { Swal } = useMessage();
-  const { onChange, setContent } = useMonaco(pomBox, {
+  const { onChange, setContent, getContent } = useMonaco(pomBox, {
     language: 'xml',
     code: props.value || defaultValue,
     options: {
@@ -106,7 +106,8 @@
     const classifierExp = /<classifier>([\s\S]*?)<\/classifier>/;
     const exclusionsExp = /<exclusions>([\s\S]*?)<\/exclusions>/;
     const invalidArtifact: Array<string> = [];
-    props.value
+    const propsValue = await getContent();
+    propsValue
       .split('</dependency>')
       .filter((x) => x.replace(/\\s+/, '') !== '')
       .forEach((dep) => {
@@ -217,8 +218,7 @@
   /* load history config records */
   async function handleReloadHistoryUploads() {
     selectedHistoryUploadJars.value = [];
-    const res = await fetchUploadJars();
-    historyUploadJars.value = res;
+    historyUploadJars.value = await fetchUploadJars();
   }
 
   const filteredHistoryUploadJarsOptions = computed(() => {
@@ -291,7 +291,7 @@
   <Tabs type="card" v-model:activeKey="activeTab" class="pom-card">
     <TabPane key="pom" tab="Maven pom">
       <div class="relative">
-        <div ref="pomBox" class="pom-box syntax-true" style="height: 300px"></div>
+        <div ref="pomBox" class="pom-box syntax-true" style="height: 330px"></div>
         <a-button type="primary" class="apply-pom" @click="handleApplyPom()">
           {{ t('common.apply') }}
         </a-button>

streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts

@@ -49,13 +49,7 @@ 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,
-  ResourceFromEnum,
-} from '/@/enums/flinkEnum';
+import { AppTypeEnum, ClusterStateEnum, ExecModeEnum, JobTypeEnum } from '/@/enums/flinkEnum';
 import { isK8sExecMode } from '../utils';
 import { useI18n } from '/@/hooks/web/useI18n';
 import { fetchCheckHadoop } from '/@/api/flink/setting';
@@ -125,21 +119,6 @@ export const useCreateAndEditSchema = (
         },
         rules: [{ required: true, message: t('flink.app.addAppTips.flinkSqlIsRequiredMessage') }],
       },
-      {
-        field: 'dependency',
-        label: t('flink.app.dependency'),
-        component: 'Input',
-        slot: 'dependency',
-        ifShow: ({ values }) => {
-          if (edit?.appId) {
-            return values.jobType == JobTypeEnum.SQL
-              ? true
-              : values.resourceFrom == ResourceFromEnum.UPLOAD;
-          } else {
-            return values?.jobType == 'sql' ? true : values?.resourceFrom != 'cvs';
-          }
-        },
-      },
       { field: 'configOverride', label: '', component: 'Input', show: false },
       {
         field: 'isSetConfig',

streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts

@@ -242,6 +242,15 @@ export const useCreateSchema = (dependencyRef: Ref) => {
           values.appType == String(AppTypeEnum.APACHE_FLINK),
         rules: [{ required: true, message: t('flink.app.addAppTips.mainClassIsRequiredMessage') }],
       },
+      {
+        field: 'dependency',
+        label: t('flink.app.dependency'),
+        component: 'Input',
+        slot: 'dependency',
+        ifShow: ({ values }) => {
+          return values?.jobType == 'sql' ? true : values?.appType == AppTypeEnum.APACHE_FLINK;
+        },
+      },
       {
         field: 'config',
         label: t('flink.app.appConf'),

streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useEditStreamPark.ts

@@ -16,7 +16,7 @@
  */
 import { FormSchema } from '/@/components/Table';
 import { computed, h, Ref, ref, unref } from 'vue';
-import { ExecModeEnum, JobTypeEnum, UseStrategyEnum } from '/@/enums/flinkEnum';
+import { AppTypeEnum, ExecModeEnum, JobTypeEnum, UseStrategyEnum } from '/@/enums/flinkEnum';
 import { useCreateAndEditSchema } from './useCreateAndEditSchema';
 import { renderSqlHistory } from './useFlinkRender';
 import { Alert } from 'ant-design-vue';
@@ -141,14 +141,24 @@ export const useEditStreamParkSchema = (
         ifShow: ({ model, values }) => values.jobType != JobTypeEnum.SQL && model.projectName,
       },
       { field: 'project', label: 'ProjectId', component: 'Input', show: false },
-
       {
         field: 'module',
         label: 'Application',
         component: 'Input',
         render: ({ model }) => h(Alert, { message: model.module, type: 'info' }),
         ifShow: ({ model, values }) => values.jobType != JobTypeEnum.SQL && model.module,
       },
+      {
+        field: 'dependency',
+        label: t('flink.app.dependency'),
+        component: 'Input',
+        slot: 'dependency',
+        ifShow: ({ values }) => {
+          return values.jobType == JobTypeEnum.SQL
+            ? true
+            : values?.appType == AppTypeEnum.APACHE_FLINK;
+        },
+      },
       { field: 'configId', label: 'configId', component: 'Input', show: false },
       { field: 'config', label: '', component: 'Input', show: false },
       { field: 'strategy', label: '', component: 'Input', show: false },

streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx

@@ -39,7 +39,7 @@ import { handleConfTemplate } from '/@/api/flink/config';
 import { decodeByBase64 } from '/@/utils/cipher';
 import { useMessage } from '/@/hooks/web/useMessage';
 import { SelectValue } from 'ant-design-vue/lib/select';
-import { CandidateTypeEnum, FailoverStrategyEnum } from '/@/enums/flinkEnum';
+import { AppTypeEnum, CandidateTypeEnum, FailoverStrategyEnum } from '/@/enums/flinkEnum';
 import { useI18n } from '/@/hooks/web/useI18n';
 import { fetchYarnQueueList } from '/@/api/flink/setting/yarnQueue';
 import { ApiSelect } from '/@/components/Form';
@@ -507,10 +507,17 @@ export const renderCompareSelectTag = (ver: any) => {
   );
 };
 
+function handleResourceFrom(model: Recordable, value: string) {
+  model.resourceFrom = value;
+  if (value == 'upload') {
+    model.appType = String(AppTypeEnum.APACHE_FLINK);
+  }
+}
+
 export const renderResourceFrom = (model: Recordable) => {
   return (
     <Select
-      onChange={(value: string) => (model.resourceFrom = value)}
+      onChange={(value: string) => handleResourceFrom(model, value)}
       value={model.resourceFrom}
       placeholder="Please select resource from"
     >

streampark-console/streampark-console-webapp/src/views/flink/app/styles/Add.less

@@ -28,9 +28,7 @@
   }
 }
 
-.sql-box,
 .pom-box {
-  height: 550px;
   width: 100%;
   overflow: hidden;
 }