去除 远程执行配置

dlink-admin/src/main/java/com/dlink/dto/StudioExecuteDTO.java

@@ -29,13 +29,11 @@ public class StudioExecuteDTO extends AbstractStatementDTO{
     private boolean statementSet;
     private boolean useSession;
     private String session;
-    private boolean useRemote;
     private Integer clusterId;
     private Integer clusterConfigurationId;
     private Integer databaseId;
     private Integer jarId;
     private boolean fragment;
-//    private String statement;
     private String jobName;
     private Integer taskId;
     private Integer maxRowNum;
@@ -61,7 +59,7 @@ public class StudioExecuteDTO extends AbstractStatementDTO{
             );
         }
         return new JobConfig(
-                type,useResult,useChangeLog,useAutoCancel, useSession, session, useRemote, clusterId,
+                type,useResult,useChangeLog,useAutoCancel, useSession, session, clusterId,
                 clusterConfigurationId,jarId, taskId, jobName, fragment,statementSet,
                 maxRowNum, checkPoint, parallelism,savePointStrategy, savePointPath,config);
     }

dlink-admin/src/main/java/com/dlink/service/impl/StudioServiceImpl.java

@@ -70,6 +70,13 @@ public class StudioServiceImpl implements StudioService {
         }
     }
 
+    private void buildSession(JobConfig config){
+        // If you are using a shared session, configure the current jobmanager address
+        if(!config.isUseSession()) {
+            config.setAddress(clusterService.buildEnvironmentAddress(config.isUseRemote(), config.getClusterId()));
+        }
+    }
+
     @Override
     public JobResult executeSql(StudioExecuteDTO studioExecuteDTO) {
         if(Dialect.isSql(studioExecuteDTO.getDialect())){
@@ -83,13 +90,7 @@ public class StudioServiceImpl implements StudioService {
     private JobResult executeFlinkSql(StudioExecuteDTO studioExecuteDTO) {
         addFlinkSQLEnv(studioExecuteDTO);
         JobConfig config = studioExecuteDTO.getJobConfig();
-        // If you are using a shared session, configure the current jobmanager address
-        if(!config.isUseSession()) {
-            if(GatewayType.LOCAL.equalsValue(config.getType())){
-                config.setUseRemote(false);
-            }
-            config.setAddress(clusterService.buildEnvironmentAddress(config.isUseRemote(), studioExecuteDTO.getClusterId()));
-        }
+        buildSession(config);
         // To initialize java udf, but it has a bug in the product environment now.
         // initUDF(config,studioExecuteDTO.getStatement());
         JobManager jobManager = JobManager.build(config);
@@ -134,9 +135,6 @@ public class StudioServiceImpl implements StudioService {
     public IResult executeDDL(StudioDDLDTO studioDDLDTO) {
         JobConfig config = studioDDLDTO.getJobConfig();
         if(!config.isUseSession()) {
-            if(GatewayType.LOCAL.equalsValue(config.getType())){
-                config.setUseRemote(false);
-            }
             config.setAddress(clusterService.buildEnvironmentAddress(config.isUseRemote(), studioDDLDTO.getClusterId()));
         }
         JobManager jobManager = JobManager.build(config);
@@ -155,12 +153,9 @@ public class StudioServiceImpl implements StudioService {
     private List<SqlExplainResult> explainFlinkSql(StudioExecuteDTO studioExecuteDTO) {
         addFlinkSQLEnv(studioExecuteDTO);
         JobConfig config = studioExecuteDTO.getJobConfig();
-        if(!config.isUseSession()) {
-            if(GatewayType.LOCAL.equalsValue(config.getType())){
-                config.setUseRemote(false);
-            }
-            config.setAddress(clusterService.buildEnvironmentAddress(config.isUseRemote(), studioExecuteDTO.getClusterId()));
-        }
+        // If you are using explainSql | getStreamGraph | getJobPlan, make the dialect change to local.
+        config.buildLocal();
+        buildSession(config);
         // To initialize java udf, but it has a bug in the product environment now.
         // initUDF(config,studioExecuteDTO.getStatement());
         JobManager jobManager = JobManager.buildPlanMode(config);
@@ -190,13 +185,9 @@ public class StudioServiceImpl implements StudioService {
     public ObjectNode getStreamGraph(StudioExecuteDTO studioExecuteDTO) {
         addFlinkSQLEnv(studioExecuteDTO);
         JobConfig config = studioExecuteDTO.getJobConfig();
-        config.setType(GatewayType.LOCAL.getLongValue());
-        if(!config.isUseSession()) {
-            if(GatewayType.LOCAL.equalsValue(config.getType())){
-                config.setUseRemote(false);
-            }
-            config.setAddress(clusterService.buildEnvironmentAddress(config.isUseRemote(), studioExecuteDTO.getClusterId()));
-        }
+        // If you are using explainSql | getStreamGraph | getJobPlan, make the dialect change to local.
+        config.buildLocal();
+        buildSession(config);
         JobManager jobManager = JobManager.buildPlanMode(config);
         return jobManager.getStreamGraph(studioExecuteDTO.getStatement());
     }
@@ -205,13 +196,9 @@ public class StudioServiceImpl implements StudioService {
     public ObjectNode getJobPlan(StudioExecuteDTO studioExecuteDTO) {
         addFlinkSQLEnv(studioExecuteDTO);
         JobConfig config = studioExecuteDTO.getJobConfig();
-        config.setType(GatewayType.LOCAL.getLongValue());
-        if(!config.isUseSession()) {
-            if(GatewayType.LOCAL.equalsValue(config.getType())){
-                config.setUseRemote(false);
-            }
-            config.setAddress(clusterService.buildEnvironmentAddress(config.isUseRemote(), studioExecuteDTO.getClusterId()));
-        }
+        // If you are using explainSql | getStreamGraph | getJobPlan, make the dialect change to local.
+        config.buildLocal();
+        buildSession(config);
         JobManager jobManager = JobManager.buildPlanMode(config);
         String planJson = jobManager.getJobPlanJson(studioExecuteDTO.getStatement());
         ObjectMapper mapper = new ObjectMapper();

dlink-admin/src/main/java/com/dlink/service/impl/TaskServiceImpl.java

@@ -285,9 +285,6 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
         JobConfig config = task.buildSubmitConfig();
         config.setJarTask(isJarTask);
         if (!JobManager.useGateway(config.getType())) {
-            if (GatewayType.LOCAL.equalsValue(config.getType())) {
-                config.setUseRemote(false);
-            }
             config.setAddress(clusterService.buildEnvironmentAddress(config.isUseRemote(), task.getClusterId()));
         } else {
             Map<String, Object> gatewayConfig = clusterConfigurationService.getGatewayConfig(task.getClusterConfigurationId());

dlink-core/src/main/java/com/dlink/job/JobConfig.java

@@ -2,6 +2,7 @@ package com.dlink.job;
 
 import com.dlink.assertion.Asserts;
 import com.dlink.executor.ExecutorSetting;
+import com.dlink.gateway.GatewayType;
 import com.dlink.gateway.config.AppConfig;
 import com.dlink.gateway.config.ClusterConfig;
 import com.dlink.gateway.config.FlinkConfig;
@@ -63,7 +64,7 @@ public class JobConfig {
         this.config = config;
     }
 
-    public JobConfig(String type, boolean useResult, boolean useChangeLog, boolean useAutoCancel, boolean useSession, String session, boolean useRemote, Integer clusterId,
+    public JobConfig(String type, boolean useResult, boolean useChangeLog, boolean useAutoCancel, boolean useSession, String session, Integer clusterId,
                      Integer clusterConfigurationId, Integer jarId, Integer taskId, String jobName, boolean useSqlFragment,
                      boolean useStatementSet, Integer maxRowNum, Integer checkpoint, Integer parallelism,
                      Integer savePointStrategyValue, String savePointPath, Map<String,String> config) {
@@ -73,7 +74,7 @@ public class JobConfig {
         this.useAutoCancel = useAutoCancel;
         this.useSession = useSession;
         this.session = session;
-        this.useRemote = useRemote;
+        this.useRemote = true;
         this.clusterId = clusterId;
         this.clusterConfigurationId = clusterConfigurationId;
         this.jarId = jarId;
@@ -213,4 +214,12 @@ public class JobConfig {
             gatewayConfig.getFlinkConfig().getConfiguration().put(entry.getKey(), (String) entry.getValue());
         }
     }
+
+    public boolean isUseRemote() {
+        return !GatewayType.LOCAL.equalsValue(type);
+    }
+
+    public void buildLocal(){
+        type = GatewayType.LOCAL.getLongValue();
+    }
 }

dlink-core/src/test/java/com/dlink/core/JobManagerTest.java

@@ -26,7 +26,7 @@ public class JobManagerTest {
     @Test
     public void cancelJobSelect(){
 
-        JobConfig config = new JobConfig("session-yarn",true,true, true,true, "s1", true, 2,
+        JobConfig config = new JobConfig("session-yarn",true,true, true,true, "s1", 2,
                 null, null,null, "测试", false,false, 100, 0,
                 1, 0,null,new HashMap<>());
         if(config.isUseRemote()) {

dlink-web/src/components/Studio/StudioRightTool/StudioConfig/index.tsx

 import {connect} from "umi";
 import {StateType} from "@/pages/FlinkSqlStudio/model";
 import {
-  Form, InputNumber,Switch, Row, Col, Tooltip, Button, Badge,
-  Typography
+  Form, InputNumber,Switch, Row, Col, Tooltip, Button,
 } from "antd";
 import {InfoCircleOutlined,MinusSquareOutlined} from "@ant-design/icons";
 import styles from "./index.less";
 import { Scrollbars } from 'react-custom-scrollbars';
 
-const { Text } = Typography;
-
 const StudioConfig = (props: any) => {
 
-  const {current,form,dispatch,tabs,currentSession,toolHeight} = props;
+  const {current,form,dispatch,tabs,toolHeight} = props;
 
   form.setFieldsValue(current.task);
 
@@ -94,16 +91,6 @@ const StudioConfig = (props: any) => {
           </Form.Item>
         </Col>
       </Row>
-      <Form.Item
-        label="远程执行" className={styles.form_item} name="useRemote" valuePropName="checked"
-        tooltip={{ title: '开启远程执行，将在远程集群进行任务执行', icon: <InfoCircleOutlined /> }}
-      >
-        {
-          currentSession.session?
-            (currentSession.sessionConfig&&currentSession.sessionConfig.useRemote?(<><Badge status="success"/><Text type="success">已启用</Text></>):(<><Badge status="error"/><Text type="danger">已禁用</Text></>)
-            ):(<Switch checkedChildren="启用" unCheckedChildren="禁用"/>)
-        }
-      </Form.Item>
     </Form>
       </Scrollbars>
       </>

dlink-web/src/pages/FlinkSqlStudio/model.ts

@@ -91,7 +91,6 @@ export type TaskType = {
   useChangeLog: boolean;
   useAutoCancel: boolean;
   useSession: boolean;
-  useRemote: boolean;
 };
 
 export type ConsoleType = {
@@ -125,7 +124,6 @@ export type SessionType = {
   session?: string;
   sessionConfig?: {
     type?: string;
-    useRemote?: boolean;
     clusterId?: number;
     clusterName?: string;
     address?: string;