[BugFix][Flink Jar]Fix FLINK JAR submission (#4073)

DataLinkDC · Dec 22, 2024 · de3aeed · de3aeed
1 parent 9cf618c
commit de3aeed
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 4 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/StudioServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/StudioServiceImpl.java
@@ -52,6 +52,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import org.springframework.stereotype.Service;
 
@@ -122,8 +123,10 @@ public LineageResult getLineage(StudioLineageDTO studioCADTO) {
             TaskDTO taskDTO = taskService.getTaskInfoById(studioCADTO.getTaskId());
             taskDTO.setStatement(taskService.buildEnvSql(taskDTO) + studioCADTO.getStatement());
             JobConfig jobConfig = taskDTO.getJobConfig();
-            jobConfig.setUdfRefer(studioCADTO.getConfigJson().getUdfReferMaps());
-            jobConfig.setConfigJson(studioCADTO.getConfigJson().getCustomConfigMaps());
+            Optional.ofNullable(studioCADTO.getConfigJson()).ifPresent(config -> {
+                jobConfig.setUdfRefer(config.getUdfReferMaps());
+                jobConfig.setConfigJson(config.getCustomConfigMaps());
+            });
 
             return LineageBuilder.getColumnLineageByLogicalPlan(taskDTO.getStatement(), jobConfig);
         }

diff --git a/dinky-core/src/main/java/org/dinky/executor/Executor.java b/dinky-core/src/main/java/org/dinky/executor/Executor.java
@@ -217,9 +217,10 @@ public JobStatementPlan parseStatementIntoJobStatementPlan(String[] statements)
                 jobStatementPlan.addJobStatement(statement, JobStatementType.SET, operationType);
             } else if (operationType.equals(SqlType.EXECUTE_JAR)) {
                 JarSubmitParam jarSubmitParam = JarSubmitParam.build(statement);
-                jarSubmitParam.setUri("base64@" + Base64.encode(pretreatStatement(jarSubmitParam.getArgs())));
+                String args = jarSubmitParam.getArgs();
+                jarSubmitParam.setUri("base64@" + Base64.encode(isUseSqlFragment() ? pretreatStatement(args) : args));
                 jobStatementPlan.addJobStatement(
-                        jarSubmitParam.toString(), JobStatementType.EXECUTE_JAR, operationType);
+                        jarSubmitParam.getStatement(), JobStatementType.EXECUTE_JAR, operationType);
             } else if (operationType.equals(SqlType.EXECUTE)) {
                 jobStatementPlan.addJobStatement(statement, JobStatementType.PIPELINE, operationType);
             } else if (operationType.equals(SqlType.PRINT)) {