修复代码无法提交到yarn中的问题

flink-yun-agent/src/main/java/com/isxcode/acorn/agent/run/KubernetesAcorn.java

@@ -85,7 +85,7 @@ public GetJobInfoRes getJobInfo(GetJobInfoReq getJobInfoReq) {
         try (KubernetesClusterDescriptor clusterDescriptor =
             kubernetesClusterClientFactory.createClusterDescriptor(flinkConfig)) {
             ClusterClientProvider<String> clusterClientProvider =
-                clusterDescriptor.retrieve(getJobInfoReq.getClusterId());
+                clusterDescriptor.retrieve(getJobInfoReq.getJobId());
             System.out.println(clusterClientProvider.getClusterClient().getWebInterfaceURL());
             return null;
         } catch (Exception e) {

flink-yun-agent/src/main/java/com/isxcode/acorn/agent/run/YarnAcorn.java

@@ -1,13 +1,12 @@
 package com.isxcode.acorn.agent.run;
 
+import com.alibaba.fastjson2.JSON;
+import com.isxcode.acorn.api.agent.pojos.dto.FlinkVerticesDto;
 import com.isxcode.acorn.api.agent.pojos.req.GetJobInfoReq;
 import com.isxcode.acorn.api.agent.pojos.req.GetJobLogReq;
 import com.isxcode.acorn.api.agent.pojos.req.StopJobReq;
 import com.isxcode.acorn.api.agent.pojos.req.SubmitJobReq;
-import com.isxcode.acorn.api.agent.pojos.res.GetJobInfoRes;
-import com.isxcode.acorn.api.agent.pojos.res.GetJobLogRes;
-import com.isxcode.acorn.api.agent.pojos.res.StopJobRes;
-import com.isxcode.acorn.api.agent.pojos.res.SubmitJobRes;
+import com.isxcode.acorn.api.agent.pojos.res.*;
 import com.isxcode.acorn.backend.api.base.exceptions.IsxAppException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -18,12 +17,27 @@
 import org.apache.flink.yarn.YarnClusterClientFactory;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.flink.yarn.configuration.YarnConfigOptions;
+import org.apache.flink.yarn.configuration.YarnConfigOptionsInternal;
 import org.apache.flink.yarn.configuration.YarnDeploymentTarget;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
+import org.springframework.web.client.RestTemplate;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static java.util.Collections.singletonList;
 
@@ -38,19 +52,17 @@ public SubmitJobRes submitJob(SubmitJobReq submitJobReq) {
         Configuration flinkConfig = GlobalConfiguration.loadConfiguration();
         flinkConfig.set(DeploymentOptions.TARGET, YarnDeploymentTarget.APPLICATION.getName());
         flinkConfig.set(PipelineOptions.NAME, submitJobReq.getAppName());
-        flinkConfig.set(ApplicationConfiguration.APPLICATION_ARGS, singletonList(""));
+        flinkConfig.set(ApplicationConfiguration.APPLICATION_ARGS, singletonList(Base64.getEncoder().encodeToString(JSON.toJSONString(submitJobReq.getAcornPluginReq()).getBytes())));
         flinkConfig.set(ApplicationConfiguration.APPLICATION_MAIN_CLASS, submitJobReq.getEntryClass());
-        flinkConfig.set(PipelineOptions.JARS, singletonList(
-            "/Users/ispong/isxcode/flink-yun/flink-yun-plugins/flink-sql-execute-plugin/build/libs/flink-sql-execute-plugin.jar"));
+        flinkConfig.set(PipelineOptions.JARS, singletonList(submitJobReq.getAppResource()));
         flinkConfig.set(DeploymentOptionsInternal.CONF_DIR, submitJobReq.getFlinkHome() + "/conf");
         flinkConfig.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("2g"));
         flinkConfig.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("2g"));
         flinkConfig.set(TaskManagerOptions.NUM_TASK_SLOTS, 1);
         flinkConfig.set(YarnConfigOptions.APPLICATION_NAME, submitJobReq.getAppName());
-        flinkConfig.set(YarnConfigOptions.FLINK_DIST_JAR,
-            "/Users/ispong/isxcode/flink-yun/flink-yun-dist/flink-min/lib/flink-dist-1.18.1.jar");
+        flinkConfig.set(YarnConfigOptions.FLINK_DIST_JAR, submitJobReq.getFlinkHome() + "/lib/flink-dist-1.18.1.jar");
         List<String> libFile = new ArrayList<>();
-        libFile.add("/Users/ispong/isxcode/flink-yun/flink-yun-dist/flink-min/lib");
+        libFile.add(submitJobReq.getFlinkHome() + "/lib");
         libFile.add("/Users/ispong/isxcode/flink-yun/resources/jdbc/system");
         libFile.add("/Users/ispong/isxcode/flink-yun/resources/cdc");
         flinkConfig.set(YarnConfigOptions.SHIP_FILES, libFile);
@@ -60,32 +72,93 @@ public SubmitJobRes submitJob(SubmitJobReq submitJobReq) {
                 .setSlotsPerTaskManager(2).createClusterSpecification();
 
         ApplicationConfiguration applicationConfiguration = ApplicationConfiguration.fromConfiguration(flinkConfig);
-        applicationConfiguration.applyToConfiguration(flinkConfig);
         YarnClusterClientFactory yarnClusterClientFactory = new YarnClusterClientFactory();
         try (YarnClusterDescriptor clusterDescriptor = yarnClusterClientFactory.createClusterDescriptor(flinkConfig)) {
             ClusterClientProvider<ApplicationId> applicationIdClusterClientProvider =
                 clusterDescriptor.deployApplicationCluster(clusterSpecification, applicationConfiguration);
-            System.out.println(applicationIdClusterClientProvider.getClusterClient().getClusterId());
-            System.out.println(applicationIdClusterClientProvider.getClusterClient().getWebInterfaceURL());
-            return null;
+            return SubmitJobRes.builder().jobId(String.valueOf(applicationIdClusterClientProvider.getClusterClient().getClusterId())).build();
         } catch (Exception e) {
-            e.printStackTrace();
             throw new IsxAppException("提交任务失败" + e.getMessage());
         }
     }
 
     @Override
     public GetJobInfoRes getJobInfo(GetJobInfoReq getJobInfoReq) {
-        return null;
+
+        Configuration flinkConfig = GlobalConfiguration.loadConfiguration();
+        flinkConfig.set(DeploymentOptionsInternal.CONF_DIR, getJobInfoReq.getFlinkHome() + "/conf");
+
+        YarnClusterClientFactory yarnClusterClientFactory = new YarnClusterClientFactory();
+        try (YarnClusterDescriptor clusterDescriptor = yarnClusterClientFactory.createClusterDescriptor(flinkConfig)) {
+            ApplicationReport applicationReport = clusterDescriptor.getYarnClient().getApplicationReport(ApplicationId.fromString(getJobInfoReq.getJobId()));
+            return GetJobInfoRes.builder().jobId(getJobInfoReq.getJobId()).status(applicationReport.getYarnApplicationState().toString()).build();
+        } catch (Exception e) {
+            throw new IsxAppException("提交任务失败" + e.getMessage());
+        }
     }
 
     @Override
     public GetJobLogRes getJobLog(GetJobLogReq getJobLogReq) {
-        return null;
+
+        String getLogCmdFormat = "yarn logs -applicationId %s";
+        Process process;
+        try {
+            process = Runtime.getRuntime().exec(String.format(getLogCmdFormat, getJobLogReq.getJobId()));
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+        InputStream inputStream = process.getInputStream();
+        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
+
+        StringBuilder errLog = new StringBuilder();
+        String line;
+        while (true) {
+            try {
+                if (!((line = reader.readLine()) != null)) break;
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+            errLog.append(line).append("\n");
+        }
+
+        try {
+            int exitCode = process.waitFor();
+            if (exitCode == 1) {
+                throw new IsxAppException(errLog.toString());
+            } else {
+                Pattern regex = Pattern.compile("LogType:taskmanager.log\\s*([\\s\\S]*?)\\s*End of LogType:taskmanager.log");
+                Matcher matcher = regex.matcher(errLog);
+                String log = "";
+                while (matcher.find()) {
+                    String tmpLog = matcher.group();
+                    if (tmpLog.contains("ERROR")) {
+                        log = tmpLog;
+                        break;
+                    }
+                    if (tmpLog.length() > log.length()) {
+                        log = tmpLog;
+                    }
+                }
+                return GetJobLogRes.builder().log(log).build();
+            }
+        } catch (InterruptedException e) {
+            throw new IsxAppException(e.getMessage());
+        }
     }
 
     @Override
     public StopJobRes stopJobReq(StopJobReq stopJobReq) {
-        return null;
+
+        Configuration flinkConfig = GlobalConfiguration.loadConfiguration();
+        flinkConfig.set(DeploymentOptionsInternal.CONF_DIR, stopJobReq.getFlinkHome() + "/conf");
+
+        YarnClusterClientFactory yarnClusterClientFactory = new YarnClusterClientFactory();
+        try (YarnClusterDescriptor clusterDescriptor = yarnClusterClientFactory.createClusterDescriptor(flinkConfig)) {
+            clusterDescriptor.getYarnClient().killApplication(ApplicationId.fromString(stopJobReq.getJobId()));
+            return StopJobRes.builder().build();
+        } catch (Exception e) {
+            throw new IsxAppException("停止任务失败" + e.getMessage());
+        }
     }
 }

flink-yun-agent/src/main/java/com/isxcode/acorn/agent/service/AcornAgentBizService.java

@@ -49,41 +49,35 @@ public GetJobInfoRes getJobInfo(GetJobInfoReq getJobInfoReq) {
 
         switch (getJobInfoReq.getAgentType()) {
             case AgentType.YARN:
-
-                break;
+                return yarnAcorn.getJobInfo(getJobInfoReq);
             case AgentType.K8S:
                 return kubernetesAcorn.getJobInfo(getJobInfoReq);
             case AgentType.FlinkCluster:
                 return flinkClusterAcorn.getJobInfo(getJobInfoReq);
             default:
                 throw new IsxAppException("agent类型不支持");
         }
-
-        return null;
     }
 
     public GetJobLogRes getJobLog(GetJobLogReq getJobLogReq) {
 
         switch (getJobLogReq.getAgentType()) {
             case AgentType.YARN:
-                break;
+                return yarnAcorn.getJobLog(getJobLogReq);
             case AgentType.K8S:
                 return kubernetesAcorn.getJobLog(getJobLogReq);
             case AgentType.FlinkCluster:
                 return flinkClusterAcorn.getJobLog(getJobLogReq);
             default:
                 throw new IsxAppException("agent类型不支持");
         }
-
-        return null;
     }
 
     public StopJobRes stopJob(StopJobReq stopJobReq) {
 
         switch (stopJobReq.getAgentType()) {
             case AgentType.YARN:
-
-                break;
+                return yarnAcorn.stopJobReq(stopJobReq);
             case AgentType.K8S:
                 return kubernetesAcorn.stopJobReq(stopJobReq);
             case AgentType.FlinkCluster:
@@ -92,6 +86,5 @@ public StopJobRes stopJob(StopJobReq stopJobReq) {
                 throw new IsxAppException("agent类型不支持");
         }
 
-        return null;
     }
 }

flink-yun-backend/flink-yun-api/src/main/java/com/isxcode/acorn/api/agent/pojos/req/GetJobInfoReq.java

@@ -16,6 +16,4 @@ public class GetJobInfoReq {
     private String flinkHome;
 
     private String agentType;
-
-    private String clusterId;
 }

flink-yun-backend/flink-yun-modules/src/main/java/com/isxcode/acorn/modules/work/run/FlinkSqlExecutor.java

@@ -150,6 +150,7 @@ protected void execute(WorkRunContext workRunContext, WorkInstanceEntity workIns
         SubmitJobReq submitJobReq = SubmitJobReq.builder().entryClass("com.isxcode.acorn.plugin.sql.execute.Job")
             .appResource(
                 "/Users/ispong/isxcode/flink-yun/flink-yun-plugins/flink-sql-execute-plugin/build/libs/flink-sql-execute-plugin.jar")
+            .appName("zhiliuyun-job")
             .acornPluginReq(AcornPluginReq.builder().flinkSql(workRunContext.getScript()).build())
             .flinkHome(engineNode.getFlinkHomePath()).agentType(calculateEngineEntityOptional.get().getClusterType())
             .build();