[Improve] support the variable in Flink configuration (#4065)

* [Improve] support the variable in Flink configuration

* [Improve] program args bug fixed.

* [Improve] modal style improvement

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

@@ -20,7 +20,6 @@
 import org.apache.streampark.common.conf.FlinkVersion;
 import org.apache.streampark.common.util.DeflaterUtils;
 import org.apache.streampark.common.util.PropertiesUtils;
-import org.apache.streampark.common.utils.CommonUtils;
 import org.apache.streampark.console.base.exception.ApiAlertException;
 import org.apache.streampark.console.base.exception.ApiDetailException;
 
@@ -38,7 +37,6 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Date;
 import java.util.Map;
-import java.util.Properties;
 
 @Getter
 @Setter
@@ -115,18 +113,6 @@ public Map<String, String> convertFlinkYamlAsMap() {
     return PropertiesUtils.loadFlinkConfYaml(flinkYamlString);
   }
 
-  @JsonIgnore
-  public Properties getFlinkConfig(Application application) {
-    String flinkYamlString = DeflaterUtils.unzipString(flinkConf);
-    Properties flinkConfig = new Properties();
-    Map<String, String> config = PropertiesUtils.loadFlinkConfYaml(flinkYamlString);
-    for (Map.Entry<String, String> entry : config.entrySet()) {
-      String value = CommonUtils.fixedValueBaseVar(entry.getValue(), application.getJobName());
-      flinkConfig.setProperty(entry.getKey(), value);
-    }
-    return flinkConfig;
-  }
-
   @JsonIgnore
   public FlinkVersion getFlinkVersion() {
     if (this.flinkVersion == null) {

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

@@ -18,12 +18,14 @@
 package org.apache.streampark.console.core.service;
 
 import org.apache.streampark.console.base.domain.RestRequest;
+import org.apache.streampark.console.core.entity.Application;
 import org.apache.streampark.console.core.entity.FlinkEnv;
 
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.service.IService;
 
 import java.io.IOException;
+import java.util.Properties;
 
 public interface FlinkEnvService extends IService<FlinkEnv> {
 
@@ -97,4 +99,6 @@ public interface FlinkEnvService extends IService<FlinkEnv> {
   void validity(Long id);
 
   IPage<FlinkEnv> findPage(FlinkEnv flinkEnv, RestRequest restRequest);
+
+  Properties getFlinkConfig(FlinkEnv flinkEnv, Application application);
 }

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

@@ -1820,7 +1820,9 @@ private Map<String, Object> getProperties(Application application, FlinkEnv flin
 
     if (ExecutionMode.isKubernetesApplicationMode(application.getExecutionMode())) {
       String archiveDir =
-          flinkEnv.getFlinkConfig(application).getProperty(JobManagerOptions.ARCHIVE_DIR.key());
+          flinkEnvService
+              .getFlinkConfig(flinkEnv, application)
+              .getProperty(JobManagerOptions.ARCHIVE_DIR.key());
       if (archiveDir != null) {
         properties.put(JobManagerOptions.ARCHIVE_DIR.key(), archiveDir);
       }

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

@@ -17,16 +17,21 @@
 
 package org.apache.streampark.console.core.service.impl;
 
+import org.apache.streampark.common.util.DeflaterUtils;
+import org.apache.streampark.common.util.PropertiesUtils;
 import org.apache.streampark.console.base.domain.RestRequest;
 import org.apache.streampark.console.base.exception.ApiAlertException;
 import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
+import org.apache.streampark.console.core.entity.Application;
 import org.apache.streampark.console.core.entity.FlinkEnv;
 import org.apache.streampark.console.core.entity.Project;
 import org.apache.streampark.console.core.mapper.FlinkEnvMapper;
 import org.apache.streampark.console.core.service.ApplicationService;
 import org.apache.streampark.console.core.service.FlinkClusterService;
 import org.apache.streampark.console.core.service.FlinkEnvService;
 
+import org.apache.commons.lang3.StringUtils;
+
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -40,6 +45,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.Date;
+import java.util.Map;
+import java.util.Properties;
 
 @Slf4j
 @Service
@@ -164,6 +171,25 @@ public IPage<FlinkEnv> findPage(FlinkEnv flinkEnv, RestRequest restRequest) {
     return this.baseMapper.findPage(page, flinkEnv);
   }
 
+  @Override
+  public Properties getFlinkConfig(FlinkEnv flinkEnv, Application application) {
+    String flinkYamlString = DeflaterUtils.unzipString(flinkEnv.getFlinkConf());
+    Properties flinkConfig = new Properties();
+    Map<String, String> config = PropertiesUtils.loadFlinkConfYaml(flinkYamlString);
+    for (Map.Entry<String, String> entry : config.entrySet()) {
+      String value = entry.getValue();
+      if (StringUtils.isNotBlank(application.getJobName())) {
+        value =
+            value.replaceAll("\\$\\{job(Name|name)}|\\$job(Name|name)", application.getJobName());
+      }
+      if (application.getId() != null) {
+        value = value.replaceAll("\\$\\{job(Id|id)}|\\$job(Id|id)", application.getId().toString());
+      }
+      flinkConfig.setProperty(entry.getKey(), value);
+    }
+    return flinkConfig;
+  }
+
   private void checkOrElseAlert(FlinkEnv flinkEnv) {
 
     // 1.check exists

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

@@ -156,7 +156,7 @@ private void expire(Savepoint entity) {
 
     if (cpThreshold == 0) {
       String flinkConfNumRetained =
-          flinkEnv.getFlinkConfig(application).getProperty(numRetainedKey);
+          flinkEnvService.getFlinkConfig(flinkEnv, application).getProperty(numRetainedKey);
       int numRetainedDefaultValue = 1;
       if (flinkConfNumRetained != null) {
         try {
@@ -293,7 +293,7 @@ public String getSavePointPath(Application appParam) throws Exception {
     if (StringUtils.isBlank(savepointPath)) {
       // flink
       FlinkEnv flinkEnv = flinkEnvService.getById(application.getVersionId());
-      Properties flinkConfig = flinkEnv.getFlinkConfig(application);
+      Properties flinkConfig = flinkEnvService.getFlinkConfig(flinkEnv, application);
       savepointPath =
           flinkConfig.getProperty(
               CheckpointingOptions.SAVEPOINT_DIRECTORY.key(),
@@ -306,10 +306,8 @@ public String getSavePointPath(Application appParam) throws Exception {
   @Override
   public String processPath(String path, String jobName, Long jobId) {
     if (StringUtils.isNotBlank(path)) {
-      return path.replaceAll("\\$job(Id|id)", jobId.toString())
-          .replaceAll("\\$\\{job(Id|id)}", jobId.toString())
-          .replaceAll("\\$job(Name|name)", jobName)
-          .replaceAll("\\$\\{job(Name|name)}", jobName);
+      return path.replaceAll("\\$\\{job(Name|name)}|\\$job(Name|name)", jobName)
+          .replaceAll("\\$\\{job(Id|id)}|\\$job(Id|id)", jobId.toString());
     }
     return path;
   }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java

@@ -119,7 +119,7 @@ private List<TrackId> getK8sWatchingApps() {
 
   public TrackId toTrackId(Application app) {
     FlinkEnv flinkEnv = flinkEnvService.getById(app.getVersionId());
-    Properties properties = flinkEnv.getFlinkConfig(app);
+    Properties properties = flinkEnvService.getFlinkConfig(flinkEnv, app);
 
     Map<String, String> dynamicProperties =
         PropertiesUtils.extractDynamicPropertiesAsJava(app.getDynamicProperties());

streampark-console/streampark-console-webapp/src/design/index.less

@@ -115,3 +115,15 @@ textarea.ant-input,
 .ant-upload.ant-upload-drag {
   border-radius: 1px !important;
 }
+
+.pop-tip {
+  display: inline-block;
+  margin-top: 5px;
+  color: darkgrey;
+}
+
+[data-theme='dark'] {
+  .pop-tip {
+    color: #666;
+  }
+}

streampark-console/streampark-console-webapp/src/design/public.less

@@ -56,8 +56,3 @@
     opacity: 0.75;
   }
 }
-
-.extra .conf-switch {
-  color: darkgrey;
-  margin-left: 8px;
-}

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

@@ -75,8 +75,8 @@
           checkedChildren: 'ON',
           unCheckedChildren: 'OFF',
         },
-        defaultValue: true,
-        afterItem: () => h('span', { class: 'conf-switch' }, t('flink.app.view.savepointTip')),
+        defaultValue: receiveData.historySavePoint && receiveData.historySavePoint.length,
+        afterItem: () => h('span', { class: 'pop-tip' }, t('flink.app.view.savepointTip')),
       },
       {
         field: 'savepointPath',
@@ -86,7 +86,7 @@
             ? 'Select'
             : 'Input',
         afterItem: () =>
-          h('span', { class: 'conf-switch' }, handleSavePointTip(receiveData.historySavePoint)),
+          h('span', { class: 'pop-tip' }, handleSavePointTip(receiveData.historySavePoint)),
         slot: 'savepoint',
         ifShow: ({ values }) => values.restoreSavepoint,
         required: true,
@@ -99,7 +99,7 @@
           checkedChildren: 'ON',
           unCheckedChildren: 'OFF',
         },
-        afterItem: () => h('span', { class: 'conf-switch' }, t('flink.app.view.ignoreRestoredTip')),
+        afterItem: () => h('span', { class: 'pop-tip' }, t('flink.app.view.ignoreRestoredTip')),
         defaultValue: false,
         ifShow: ({ values }) => values.restoreSavepoint,
       },

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

@@ -54,9 +54,8 @@
           checkedChildren: 'ON',
           unCheckedChildren: 'OFF',
         },
-        defaultValue: true,
-        afterItem: () =>
-          h('span', { class: 'conf-switch' }, t('flink.app.operation.enableSavePoint')),
+        defaultValue: false,
+        afterItem: () => h('span', { class: 'pop-tip' }, t('flink.app.operation.enableSavePoint')),
       },
       {
         field: 'customSavepoint',
@@ -78,7 +77,7 @@
         },
         defaultValue: false,
         ifShow: ({ values }) => !!values.triggerSavepoint,
-        afterItem: () => h('span', { class: 'conf-switch' }, t('flink.app.operation.enableDrain')),
+        afterItem: () => h('span', { class: 'pop-tip' }, t('flink.app.operation.enableDrain')),
       },
     ],
     colon: true,

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

@@ -226,11 +226,7 @@ export const useFlinkApplication = (openStartModal: Fn) => {
       ],
       content: () => {
         return (
-          <Form
-            class="!pt-40px"
-            layout='vertical'
-            baseColProps = {{ span: 20, offset: 2 }}
-          >
+          <Form class="!pt-40px" layout="vertical" baseColProps={{ span: 20, offset: 2 }}>
             <Form.Item
               label="Job Name"
               validateStatus={unref(validateStatus)}
@@ -321,8 +317,8 @@ export const useFlinkApplication = (openStartModal: Fn) => {
             class="!pt-40px"
             ref={mappingRef}
             name="mappingForm"
-            baseColProps = {{ span: 20, offset: 2 }}
-            layout='vertical'
+            baseColProps={{ span: 20, offset: 2 }}
+            layout="vertical"
             v-model:model={formValue}
           >
             <Form.Item label="Job Name">

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

@@ -232,7 +232,7 @@ export const renderOptionsItems = (
             rules={[{ validator: conf.validator }]}
           />
         )}
-        {conf.type === 'switch' && <span class="conf-switch">({conf.placeholder})</span>}
+        {conf.type === 'switch' && <span>({conf.placeholder})</span>}
         <p class="conf-desc"> {descriptionFilter(conf)} </p>
       </Form.Item>
     );

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

@@ -75,6 +75,7 @@
   .conf-desc {
     color: darkgrey;
     margin-bottom: 0;
+    margin-top: 5px;
   }
 
   .sql-desc {

streampark-console/streampark-console-webapp/src/views/flink/variable/components/VariableModal.vue

@@ -120,7 +120,7 @@
         },
         defaultValue: false,
         afterItem: () =>
-          h('span', { class: 'conf-switch' }, t('flink.variable.form.desensitizationDesc')),
+          h('span', { class: 'pop-tip' }, t('flink.variable.form.desensitizationDesc')),
       },
       {
         field: 'description',
@@ -188,11 +188,3 @@
     }
   }
 </script>
-
-<style lang="less">
-  .conf-switch {
-    display: inline-block;
-    margin-top: 10px;
-    color: darkgrey;
-  }
-</style>

streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertModal.vue

@@ -56,7 +56,7 @@
           allowClear: true,
           placeholder: t('setting.alarm.alertNamePlaceHolder'),
         },
-        afterItem: () => h('span', { class: 'conf-switch' }, t('setting.alarm.alertNameTips')),
+        afterItem: () => h('span', { class: 'pop-tip' }, t('setting.alarm.alertNameTips')),
         dynamicRules: () => {
           return [
             {

streampark-console/streampark-console-webapp/src/views/setting/ExternalLink/components/Modal.vue

@@ -80,7 +80,7 @@
         afterItem: () =>
           h(
             'span',
-            { class: 'conf-switch' },
+            { class: 'pop-tip' },
             'Supported variables: {job_id}, {yarn_id}, {job_name}, Example: https://grafana/flink-monitoring?var-JobId=var-JobId={job_id}',
           ),
         rules: [

streampark-console/streampark-console-webapp/src/views/setting/FlinkHome/components/Modal.vue

@@ -42,7 +42,7 @@
           allowClear: true,
         },
         afterItem: () =>
-          h('span', { class: 'conf-switch' }, t('setting.flinkHome.operateMessage.flinkNameTips')),
+          h('span', { class: 'pop-tip' }, t('setting.flinkHome.operateMessage.flinkNameTips')),
         rules: [
           { required: true, message: t('setting.flinkHome.operateMessage.flinkNameIsRequired') },
         ],
@@ -56,7 +56,7 @@
           allowClear: true,
         },
         afterItem: () =>
-          h('span', { class: 'conf-switch' }, t('setting.flinkHome.operateMessage.flinkHomeTips')),
+          h('span', { class: 'pop-tip' }, t('setting.flinkHome.operateMessage.flinkHomeTips')),
         rules: [
           { required: true, message: t('setting.flinkHome.operateMessage.flinkHomeIsRequired') },
         ],
@@ -172,10 +172,3 @@
     </div>
   </BasicModal>
 </template>
-<style lang="less">
-  .conf-switch {
-    display: inline-block;
-    margin-top: 10px;
-    color: darkgrey;
-  }
-</style>