[Optimize]Optimize some code (#3831)

Co-authored-by: zackyoungh <zackyoungh@users.noreply.github.com>
DataLinkDC · Sep 24, 2024 · 81db6e4 · 81db6e4
1 parent 0f52cb2
commit 81db6e4
Show file tree

Hide file tree

Showing 20 changed files with 221 additions and 82 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/aop/LogAspect.java b/dinky-admin/src/main/java/org/dinky/aop/LogAspect.java
@@ -103,6 +103,9 @@ protected void handleCommonLogic(final JoinPoint joinPoint, final Exception e, O
             // *========数据库日志=========*//
             OperateLog operLog = new OperateLog();
             Result<Void> result = JsonUtils.toBean(jsonResult, new TypeReference<Result<Void>>() {});
+            if (result == null) {
+                result = Result.failed();
+            }
             operLog.setStatus(result.isSuccess() ? BusinessStatus.SUCCESS.ordinal() : BusinessStatus.FAIL.ordinal());
 
             // 请求的地址

diff --git a/dinky-common/src/main/java/org/dinky/context/FlinkUdfPathContextHolder.java b/dinky-common/src/main/java/org/dinky/context/FlinkUdfPathContextHolder.java
@@ -20,17 +20,21 @@
 package org.dinky.context;
 
 import java.io.File;
+import java.util.Arrays;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * @since 0.7.0
  */
 public class FlinkUdfPathContextHolder {
+    private static final List<String> PYTHON_FILE_SUFFIX =
+            Arrays.asList(".zip", ".py", ".pyc", ".pyo", ".pyd", ".pyw", ".pyz", ".pyzw");
 
     private final Set<File> UDF_PATH_CONTEXT = new HashSet<>();
     private final Set<File> OTHER_PLUGINS_PATH_CONTEXT = new HashSet<>();
-    private final Set<File> PYTHON_UDF_FILE = new HashSet<>();
     private final Set<File> FILES = new HashSet<>();
 
     public void addUdfPath(File file) {
@@ -54,7 +58,10 @@ public Set<File> getUdfFile() {
     }
 
     public Set<File> getPyUdfFile() {
-        return PYTHON_UDF_FILE;
+        return getAllFileSet().stream()
+                .filter(file -> PYTHON_FILE_SUFFIX.stream()
+                        .anyMatch(suffix -> file.getName().endsWith(suffix)))
+                .collect(Collectors.toSet());
     }
 
     public Set<File> getOtherPluginsFiles() {

diff --git a/dinky-core/src/main/java/org/dinky/data/result/SelectResult.java b/dinky-core/src/main/java/org/dinky/data/result/SelectResult.java
@@ -31,6 +31,7 @@
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.collection.ListUtil;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
@@ -42,6 +43,7 @@
 @Slf4j
 @Setter
 @Getter
+@NoArgsConstructor
 public class SelectResult extends AbstractResult implements IResult {
 
     private String jobID;
@@ -65,7 +67,6 @@ public SelectResult(
         this.columns = columns;
         this.jobID = jobID;
         this.success = success;
-        // this.endTime = LocalDateTime.now();
         this.isDestroyed = false;
     }
 

diff --git a/dinky-gateway/src/main/java/org/dinky/gateway/yarn/YarnGateway.java b/dinky-gateway/src/main/java/org/dinky/gateway/yarn/YarnGateway.java
@@ -112,6 +112,7 @@ public YarnGateway(GatewayConfig config) {
         super(config);
     }
 
+    @Override
     public void init() {
         initConfig();
         initYarnClient();
@@ -192,7 +193,7 @@ private void initYarnClient() {
                 hadoopUserName = "hdfs";
             }
 
-            // 设置 yarn 提交的用户名
+            // Set the username for the yarn submission
             String yarnUser = configuration.get(CustomerConfigureOptions.YARN_APPLICATION_USER);
             if (StrUtil.isNotBlank(yarnUser)) {
                 UserGroupInformation.setLoginUser(UserGroupInformation.createRemoteUser(yarnUser));
@@ -211,6 +212,7 @@ private Path getYanConfigFilePath(String path) {
         return new Path(URI.create(config.getClusterConfig().getHadoopConfigPath() + "/" + path));
     }
 
+    @Override
     public SavePointResult savepointCluster(String savePoint) {
         if (Asserts.isNull(yarnClient)) {
             init();
@@ -221,6 +223,7 @@ public SavePointResult savepointCluster(String savePoint) {
         return runClusterSavePointResult(savePoint, applicationId, clusterDescriptor);
     }
 
+    @Override
     public SavePointResult savepointJob(String savePoint) {
         if (Asserts.isNull(yarnClient)) {
             init();
@@ -253,13 +256,14 @@ private void autoCancelCluster(ClusterClient<ApplicationId> clusterClient) {
                 Thread.sleep(3000);
                 clusterClient.shutDownCluster();
             } catch (InterruptedException e) {
-                e.printStackTrace();
+                logger.error(e.getMessage());
             } finally {
                 clusterClient.close();
             }
         });
     }
 
+    @Override
     public TestResult test() {
         try {
             initConfig();
@@ -365,13 +369,12 @@ protected YarnClusterDescriptor createYarnClusterDescriptorWithJar(FlinkUdfPathC
     }
 
     protected YarnClusterDescriptor createInitYarnClusterDescriptor() {
-        YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(
+        return new YarnClusterDescriptor(
                 configuration,
                 yarnConfiguration,
                 yarnClient,
                 YarnClientYarnClusterInformationRetriever.create(yarnClient),
                 true);
-        return yarnClusterDescriptor;
     }
 
     protected String getWebUrl(ClusterClient<ApplicationId> clusterClient, YarnResult result)
@@ -464,7 +467,7 @@ public String getLatestJobManageHost(String appId, String oldJobManagerHost) {
         HighAvailabilityMode highAvailabilityMode = HighAvailabilityMode.fromConfig(configuration);
 
         if (HighAvailabilityMode.ZOOKEEPER == highAvailabilityMode) {
-            configuration.setString(HighAvailabilityOptions.HA_CLUSTER_ID, appId);
+            configuration.set(HighAvailabilityOptions.HA_CLUSTER_ID, appId);
             String zkQuorum = configuration.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM);
 
             if (zkQuorum == null || StringUtils.isBlank(zkQuorum)) {
@@ -495,13 +498,13 @@ public String getLatestJobManageHost(String appId, String oldJobManagerHost) {
                     }
                 }
             } catch (Exception e) {
-                e.printStackTrace();
+                logger.error("", e);
             } finally {
                 if (Asserts.isNotNull(zooKeeper)) {
                     try {
                         zooKeeper.close();
                     } catch (InterruptedException e) {
-                        e.printStackTrace();
+                        logger.error("", e);
                     }
                 }
             }
@@ -515,12 +518,11 @@ public String getLatestJobManageHost(String appId, String oldJobManagerHost) {
      * Creates a ZooKeeper path of the form "/a/b/.../z".
      */
     private static String generateZookeeperPath(String... paths) {
-        final String result = Arrays.stream(paths)
+
+        return Arrays.stream(paths)
                 .map(YarnGateway::trimSlashes)
                 .filter(s -> !s.isEmpty())
                 .collect(Collectors.joining("/", "/", ""));
-
-        return result;
     }
 
     private static String trimSlashes(String input) {

diff --git a/dinky-web/src/components/Flink/FlinkDag/index.tsx b/dinky-web/src/components/Flink/FlinkDag/index.tsx
@@ -168,9 +168,10 @@ function getMaxWidthAndDepth(edges: CusEdge[]): { maxWidth: number; maxDepth: nu
     graph[sourceCell].push(targetCell);
   });
 
-  const maxSource = Object.keys(sourceCount).reduce((a, b) =>
-    sourceCount[a] > sourceCount[b] ? a : b
-  );
+  const maxSource =
+    Object.keys(sourceCount).length < 1
+      ? '1'
+      : Object.keys(sourceCount).reduce((a, b) => (sourceCount[a] > sourceCount[b] ? a : b));
   const maxWidth = sourceCount[maxSource];
 
   const visited: Record<string, boolean> = {};

diff --git a/dinky-web/src/components/Flink/OptionsSelect/index.tsx b/dinky-web/src/components/Flink/OptionsSelect/index.tsx
@@ -41,7 +41,7 @@ const FlinkOptionsSelect = (props: FlinkOptionsProps) => {
   return (
     <ProFormSelect
       {...props}
-      fieldProps={{ dropdownRender: (item) => renderTemplateDropDown(item) }}
+      fieldProps={{ dropdownRender: (item) => renderTemplateDropDown(item), virtual: false }}
     />
   );
 };