[Optimization-3959][core] Optimize submit task print error log (#3960)

aiwenmo · web-flow · commit d64de82913b8 · 2024-11-27T22:49:29.000+08:00
diff --git a/dinky-admin/pom.xml b/dinky-admin/pom.xml
@@ -359,10 +359,26 @@
             <groupId>org.dinky</groupId>
             <artifactId>dinky-alert-dingtalk</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.dinky</groupId>
+            <artifactId>dinky-alert-feishu</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.dinky</groupId>
+            <artifactId>dinky-alert-wechat</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.dinky</groupId>
             <artifactId>dinky-alert-sms</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.dinky</groupId>
+            <artifactId>dinky-alert-email</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.dinky</groupId>
+            <artifactId>dinky-alert-http</artifactId>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/dinky-admin/src/main/java/org/dinky/init/SystemInit.java b/dinky-admin/src/main/java/org/dinky/init/SystemInit.java
@@ -31,6 +31,7 @@
 import org.dinky.data.model.Task;
 import org.dinky.data.model.job.JobInstance;
 import org.dinky.data.model.rbac.Tenant;
+import org.dinky.function.FlinkUDFDiscover;
 import org.dinky.function.constant.PathConstant;
 import org.dinky.function.pool.UdfCodePool;
 import org.dinky.job.ClearJobHistoryTask;
@@ -106,6 +107,7 @@ public void run(ApplicationArguments args) {
             initDaemon();
             initDolphinScheduler();
             registerUDF();
+            discoverUDF();
             updateGitBuildState();
             registerURL();
         } catch (NoClassDefFoundError e) {
@@ -220,6 +222,10 @@ public void registerUDF() {
         UdfCodePool.updateGitPool(gitProjectService.getGitPool());
     }
 
+    public void discoverUDF() {
+        FlinkUDFDiscover.getCustomStaticUDFs();
+    }
+
     public void updateGitBuildState() {
         String path = PathConstant.TMP_PATH + "/build.list";
         if (FileUtil.exist(path)) {
diff --git a/dinky-admin/src/main/java/org/dinky/service/catalogue/factory/CatalogueTreeSortFactory.java b/dinky-admin/src/main/java/org/dinky/service/catalogue/factory/CatalogueTreeSortFactory.java
@@ -55,7 +55,6 @@ public CatalogueTreeSortStrategy getStrategy(String strategyName) {
         CatalogueTreeSortStrategy catalogueTreeSortStrategy =
                 Safes.of(catalogueTreeSortStrategyMap).get(strategyName);
         if (Objects.isNull(catalogueTreeSortStrategy)) {
-            log.warn("Strategy {} is not defined. Use DefaultStrategy", strategyName);
             catalogueTreeSortStrategy =
                     Safes.of(catalogueTreeSortStrategyMap).get(CatalogueSortConstant.STRATEGY_DEFAULT);
         }
diff --git a/dinky-admin/src/main/java/org/dinky/service/catalogue/impl/CatalogueServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/catalogue/impl/CatalogueServiceImpl.java
@@ -121,7 +121,6 @@ public class CatalogueServiceImpl extends SuperServiceImpl<CatalogueMapper, Cata
      */
     @Override
     public List<Catalogue> getCatalogueTree(CatalogueTreeQueryDTO catalogueTreeQueryDto) {
-        log.info("getCatalogueTree, catalogueTreeQueryDto: {}", catalogueTreeQueryDto);
         List<Catalogue> catalogueTree = buildCatalogueTree(this.list());
         // sort
         CatalogueTreeSortStrategy strategy = catalogueTreeSortFactory.getStrategy(catalogueTreeQueryDto.getSortValue());
diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/UDFServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/UDFServiceImpl.java
@@ -24,11 +24,11 @@
 import org.dinky.data.model.udf.UDFManage;
 import org.dinky.data.vo.CascaderVO;
 import org.dinky.data.vo.UDFManageVO;
+import org.dinky.function.FlinkUDFDiscover;
 import org.dinky.function.data.model.UDF;
 import org.dinky.mapper.UDFManageMapper;
 import org.dinky.service.UDFService;
 import org.dinky.service.resource.ResourcesService;
-import org.dinky.trans.Operations;
 import org.dinky.utils.UDFUtils;
 
 import org.apache.flink.table.catalog.FunctionLanguage;
@@ -182,7 +182,7 @@ public List<UDFManage> getUDFFromUdfManage() {
     @Override
     public List<CascaderVO> getAllUdfsToCascader(List<UDF> userDefinedReleaseUdfs) {
         // Get all UDFs of static UDFs and dynamic UDFs
-        List<UDF> staticUdfs = Operations.getCustomStaticUdfs();
+        List<UDF> staticUdfs = FlinkUDFDiscover.getCustomStaticUDFs();
 
         // get all UDFs of UDFManage table
         List<UDF> udfManageDynamic = getUDFFromUdfManage().stream()
diff --git a/dinky-app/dinky-app-1.20/pom.xml b/dinky-app/dinky-app-1.20/pom.xml
@@ -60,6 +60,7 @@
                         </manifest>
                     </archive>
                     <outputDirectory>${project.parent.parent.basedir}/build/extends</outputDirectory>
+
                 </configuration>
                 <executions>
                     <execution>
diff --git a/dinky-app/dinky-app-base/pom.xml b/dinky-app/dinky-app-base/pom.xml
@@ -46,10 +46,6 @@
             <groupId>org.dinky</groupId>
             <artifactId>dinky-core</artifactId>
             <exclusions>
-                <exclusion>
-                    <groupId>com.github.xiaoymin</groupId>
-                    <artifactId>knife4j-openapi2-spring-boot-starter</artifactId>
-                </exclusion>
                 <exclusion>
                     <groupId>com.github.docker-java</groupId>
                     <artifactId>docker-java-core</artifactId>
@@ -82,14 +78,6 @@
                     <groupId>org.dinky</groupId>
                     <artifactId>dinky-gateway</artifactId>
                 </exclusion>
-                <exclusion>
-                    <groupId>org.dinky</groupId>
-                    <artifactId>dinky-alert-sms</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.dinky</groupId>
-                    <artifactId>dinky-alert-dingtalk</artifactId>
-                </exclusion>
             </exclusions>
         </dependency>
         <dependency>
diff --git a/dinky-assembly/src/main/assembly/package.xml b/dinky-assembly/src/main/assembly/package.xml
@@ -122,6 +122,14 @@
                 <include>dinky-client-1.19-${project.version}.jar</include>
             </includes>
         </fileSet>
+        <fileSet>
+            <directory>${project.parent.basedir}/build/extends/</directory>
+            <outputDirectory>extends/flink1.20/dinky</outputDirectory>
+            <includes>
+                <include>dinky-catalog-mysql-1.20-${project.version}.jar</include>
+                <include>dinky-client-1.20-${project.version}.jar</include>
+            </includes>
+        </fileSet>
         <fileSet>
             <directory>${project.parent.basedir}/build/extends/</directory>
             <outputDirectory>extends</outputDirectory>
diff --git a/dinky-core/pom.xml b/dinky-core/pom.xml
@@ -32,11 +32,6 @@
         <hadoop.version>3.1.0</hadoop.version>
     </properties>
     <dependencies>
-        <!--<dependency>
-            <groupId>org.codehaus.groovy</groupId>
-            <artifactId>groovy</artifactId>
-            <version>3.0.9</version>
-        </dependency>-->
         <dependency>
             <groupId>org.dinky</groupId>
             <artifactId>dinky-common</artifactId>
@@ -54,36 +49,6 @@
             <artifactId>dinky-metadata-base</artifactId>
             <scope>${scope.runtime}</scope>
         </dependency>
-        <dependency>
-            <groupId>org.dinky</groupId>
-            <artifactId>dinky-alert-dingtalk</artifactId>
-            <scope>${scope.runtime}</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.dinky</groupId>
-            <artifactId>dinky-alert-wechat</artifactId>
-            <scope>${scope.runtime}</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.dinky</groupId>
-            <artifactId>dinky-alert-feishu</artifactId>
-            <scope>${scope.runtime}</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.dinky</groupId>
-            <artifactId>dinky-alert-sms</artifactId>
-            <scope>${scope.runtime}</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.dinky</groupId>
-            <artifactId>dinky-alert-email</artifactId>
-            <scope>${scope.runtime}</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.dinky</groupId>
-            <artifactId>dinky-alert-http</artifactId>
-            <scope>${scope.runtime}</scope>
-        </dependency>
         <dependency>
             <groupId>org.dinky</groupId>
             <artifactId>dinky-metadata-paimon</artifactId>
@@ -203,10 +168,6 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.reflections</groupId>
-            <artifactId>reflections</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-log4j12</artifactId>
diff --git a/dinky-core/src/main/java/org/dinky/data/result/MockResultRunnable.java b/dinky-core/src/main/java/org/dinky/data/result/MockResultRunnable.java
@@ -122,7 +122,7 @@ private void catchChangLog(SelectResult selectResult) {
                 for (String tableIdentifier : tableIdentifierList) {
                     if (!tableIdentifierIndexMap.containsKey(tableIdentifier)) {
                         tableIdentifierIndexMap.put(tableIdentifier, 0);
-                    } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {
+                    } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum - 1) {
                         allSinkFinished = true;
                         continue;
                     }
@@ -145,7 +145,7 @@ private void catchChangLog(SelectResult selectResult) {
                             rows.add(rowDataWithTableIdentifier);
                             tableIdentifierIndexMap.put(
                                     tableIdentifier, tableIdentifierIndexMap.get(tableIdentifier) + 1);
-                            if (tableIdentifierIndexMap.get(tableIdentifier) > maxRowNum) {
+                            if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {
                                 break;
                             }
                         }
@@ -190,7 +190,7 @@ private void catchData(SelectResult selectResult) {
                 for (String tableIdentifier : tableIdentifierList) {
                     if (!tableIdentifierIndexMap.containsKey(tableIdentifier)) {
                         tableIdentifierIndexMap.put(tableIdentifier, 0);
-                    } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {
+                    } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum - 1) {
                         allSinkFinished = true;
                         continue;
                     }
@@ -219,7 +219,7 @@ private void catchData(SelectResult selectResult) {
                             }
                             tableIdentifierIndexMap.put(
                                     tableIdentifier, tableIdentifierIndexMap.get(tableIdentifier) + 1);
-                            if (tableIdentifierIndexMap.get(tableIdentifier) > maxRowNum) {
+                            if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {
                                 break;
                             }
                         }
diff --git a/dinky-core/src/main/java/org/dinky/explainer/Explainer.java b/dinky-core/src/main/java/org/dinky/explainer/Explainer.java
@@ -41,6 +41,7 @@
 import org.dinky.job.builder.JobUDFBuilder;
 import org.dinky.trans.Operations;
 import org.dinky.utils.DinkyClassLoaderUtil;
+import org.dinky.utils.LogUtil;
 import org.dinky.utils.SqlUtil;
 
 import org.apache.flink.runtime.rest.messages.JobPlanInfo;
@@ -141,8 +142,9 @@ public ExplainResult explainSql(String statement) {
             jobStatementPlan.buildFinalStatement();
             jobManager.setJobStatementPlan(jobStatementPlan);
         } catch (Exception e) {
+            String error = LogUtil.getError("Exception in parsing FlinkSQL:\n" + SqlUtil.addLineNumber(statement), e);
             SqlExplainResult.Builder resultBuilder = SqlExplainResult.Builder.newBuilder();
-            resultBuilder.error(e.getMessage()).parseTrue(false);
+            resultBuilder.error(error).parseTrue(false);
             sqlExplainRecords.add(resultBuilder.build());
             log.error("Failed parseStatements:", e);
             return new ExplainResult(false, sqlExplainRecords.size(), sqlExplainRecords);
diff --git a/dinky-core/src/main/java/org/dinky/job/JobManager.java b/dinky-core/src/main/java/org/dinky/job/JobManager.java
@@ -92,7 +92,6 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
 import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.text.StrFormatter;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 
@@ -315,16 +314,14 @@ public JobResult executeSql(String statement) throws Exception {
             }
         } catch (Exception e) {
             String errorMessage = e.getMessage();
-            if (errorMessage != null && errorMessage.contains("Only insert statement is supported now")) {
-                throw new BusException(Status.OPERATE_NOT_SUPPORT_QUERY.getMessage());
-            }
-            String error = StrFormatter.format(
-                    "Exception in executing FlinkSQL:\n{}\n{}", SqlUtil.addLineNumber(currentSql), errorMessage);
             job.setEndTime(LocalDateTime.now());
             job.setStatus(Job.JobStatus.FAILED);
-            job.setError(error);
+            job.setError(errorMessage);
             failed();
-            throw new Exception(error, e);
+            if (errorMessage != null && errorMessage.contains("Only insert statement is supported now")) {
+                throw new BusException(Status.OPERATE_NOT_SUPPORT_QUERY.getMessage());
+            }
+            throw new Exception(errorMessage, e);
         } finally {
             close();
         }
diff --git a/dinky-core/src/main/java/org/dinky/job/runner/JobDDLRunner.java b/dinky-core/src/main/java/org/dinky/job/runner/JobDDLRunner.java
@@ -47,7 +47,6 @@
 import java.util.Set;
 
 import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.text.StrFormatter;
 import cn.hutool.core.util.ArrayUtil;
 import cn.hutool.core.util.RandomUtil;
 import cn.hutool.core.util.StrUtil;
@@ -125,10 +124,8 @@ public SqlExplainResult explain(JobStatement jobStatement) {
                     .sql(jobStatement.getStatement())
                     .index(jobStatement.getIndex());
         } catch (Exception e) {
-            String error = StrFormatter.format(
-                    "Exception in explaining FlinkSQL:\n{}\n{}",
-                    SqlUtil.addLineNumber(jobStatement.getStatement()),
-                    LogUtil.getError(e));
+            String error = LogUtil.getError(
+                    "Exception in explaining FlinkSQL:\n" + SqlUtil.addLineNumber(jobStatement.getStatement()), e);
             resultBuilder
                     .error(error)
                     .explainTrue(false)
diff --git a/dinky-core/src/main/java/org/dinky/job/runner/JobPipelineRunner.java b/dinky-core/src/main/java/org/dinky/job/runner/JobPipelineRunner.java
@@ -51,7 +51,6 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import cn.hutool.core.text.StrFormatter;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
@@ -100,10 +99,8 @@ public SqlExplainResult explain(JobStatement jobStatement) {
                         .explainTime(LocalDateTime.now())
                         .index(jobStatement.getIndex());
             } catch (Exception e) {
-                String error = StrFormatter.format(
-                        "Exception in explaining FlinkSQL:\n{}\n{}",
-                        SqlUtil.addLineNumber(jobStatement.getStatement()),
-                        LogUtil.getError(e));
+                String error = LogUtil.getError(
+                        "Exception in explaining FlinkSQL:\n" + SqlUtil.addLineNumber(jobStatement.getStatement()), e);
                 resultBuilder
                         .parseTrue(false)
                         .error(error)
diff --git a/dinky-core/src/main/java/org/dinky/job/runner/JobSetRunner.java b/dinky-core/src/main/java/org/dinky/job/runner/JobSetRunner.java
@@ -29,7 +29,6 @@
 
 import java.time.LocalDateTime;
 
-import cn.hutool.core.text.StrFormatter;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
@@ -66,10 +65,8 @@ public SqlExplainResult explain(JobStatement jobStatement) {
                     .explainTime(LocalDateTime.now())
                     .index(jobStatement.getIndex());
         } catch (Exception e) {
-            String error = StrFormatter.format(
-                    "Exception in explaining FlinkSQL:\n{}\n{}",
-                    SqlUtil.addLineNumber(jobStatement.getStatement()),
-                    LogUtil.getError(e));
+            String error = LogUtil.getError(
+                    "Exception in explaining FlinkSQL:\n" + SqlUtil.addLineNumber(jobStatement.getStatement()), e);
             resultBuilder
                     .parseTrue(false)
                     .error(error)
diff --git a/dinky-core/src/main/java/org/dinky/job/runner/JobSqlRunner.java b/dinky-core/src/main/java/org/dinky/job/runner/JobSqlRunner.java
diff --git a/dinky-core/src/main/java/org/dinky/trans/Operations.java b/dinky-core/src/main/java/org/dinky/trans/Operations.java
diff --git a/dinky-function/pom.xml b/dinky-function/pom.xml
diff --git a/dinky-function/src/main/java/org/dinky/function/FlinkUDFDiscover.java b/dinky-function/src/main/java/org/dinky/function/FlinkUDFDiscover.java
diff --git a/dinky-gateway/pom.xml b/dinky-gateway/pom.xml
diff --git a/dinky-metadata/dinky-metadata-paimon/pom.xml b/dinky-metadata/dinky-metadata-paimon/pom.xml
diff --git a/pom.xml b/pom.xml

Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,6 @@ public CatalogueTreeSortStrategy getStrategy(String strategyName) {`
`55`	`55`	`CatalogueTreeSortStrategy catalogueTreeSortStrategy =`
`56`	`56`	`Safes.of(catalogueTreeSortStrategyMap).get(strategyName);`
`57`	`57`	`if (Objects.isNull(catalogueTreeSortStrategy)) {`
`58`		`- log.warn("Strategy {} is not defined. Use DefaultStrategy", strategyName);`
`59`	`58`	`catalogueTreeSortStrategy =`
`60`	`59`	`Safes.of(catalogueTreeSortStrategyMap).get(CatalogueSortConstant.STRATEGY_DEFAULT);`
`61`	`60`	`}`
Original file line number	Diff line number	Diff line change
`@@ -122,7 +122,7 @@ private void catchChangLog(SelectResult selectResult) {`
`122`	`122`	`for (String tableIdentifier : tableIdentifierList) {`
`123`	`123`	`if (!tableIdentifierIndexMap.containsKey(tableIdentifier)) {`
`124`	`124`	`tableIdentifierIndexMap.put(tableIdentifier, 0);`
`125`		`- } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {`
	`125`	`+ } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum - 1) {`
`126`	`126`	`allSinkFinished = true;`
`127`	`127`	`continue;`
`128`	`128`	`}`
`@@ -145,7 +145,7 @@ private void catchChangLog(SelectResult selectResult) {`
`145`	`145`	`rows.add(rowDataWithTableIdentifier);`
`146`	`146`	`tableIdentifierIndexMap.put(`
`147`	`147`	`tableIdentifier, tableIdentifierIndexMap.get(tableIdentifier) + 1);`
`148`		`- if (tableIdentifierIndexMap.get(tableIdentifier) > maxRowNum) {`
	`148`	`+ if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {`
`149`	`149`	`break;`
`150`	`150`	`}`
`151`	`151`	`}`
`@@ -190,7 +190,7 @@ private void catchData(SelectResult selectResult) {`
`190`	`190`	`for (String tableIdentifier : tableIdentifierList) {`
`191`	`191`	`if (!tableIdentifierIndexMap.containsKey(tableIdentifier)) {`
`192`	`192`	`tableIdentifierIndexMap.put(tableIdentifier, 0);`
`193`		`- } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {`
	`193`	`+ } else if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum - 1) {`
`194`	`194`	`allSinkFinished = true;`
`195`	`195`	`continue;`
`196`	`196`	`}`
`@@ -219,7 +219,7 @@ private void catchData(SelectResult selectResult) {`
`219`	`219`	`}`
`220`	`220`	`tableIdentifierIndexMap.put(`
`221`	`221`	`tableIdentifier, tableIdentifierIndexMap.get(tableIdentifier) + 1);`
`222`		`- if (tableIdentifierIndexMap.get(tableIdentifier) > maxRowNum) {`
	`222`	`+ if (tableIdentifierIndexMap.get(tableIdentifier) >= maxRowNum) {`
`223`	`223`	`break;`
`224`	`224`	`}`
`225`	`225`	`}`