opt

Author: todd5167

launcher/src/main/java/com/dtstack/flink/sql/launcher/factory/YarnClusterClientFactory.java

@@ -23,14 +23,15 @@
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.runtime.security.SecurityConfiguration;
-import org.apache.flink.runtime.security.SecurityUtils;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.function.FunctionUtils;
 import org.apache.flink.yarn.YarnClientYarnClusterInformationRetriever;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 
 import java.io.File;
+import java.io.IOException;
 import java.util.Iterator;
 import java.util.Map;
 
@@ -41,6 +42,7 @@
  * @author maqi
  */
 public class YarnClusterClientFactory extends AbstractClusterClientFactory {
+    private static final String XML_FILE_EXTENSION = "xml";
     @Override
     public ClusterDescriptor createClusterDescriptor(String yarnConfDir, Configuration flinkConfig) {
 
@@ -70,27 +72,12 @@ public ClusterDescriptor createClusterDescriptor(String yarnConfDir, Configurati
         }
     }
 
-    private YarnConfiguration getYarnConf(String yarnConfDir) {
+    private YarnConfiguration getYarnConf(String yarnConfDir) throws IOException {
         YarnConfiguration yarnConf = new YarnConfiguration();
-        try {
-            File dir = new File(yarnConfDir);
-            if (dir.exists() && dir.isDirectory()) {
-                File[] xmlFileList = new File(yarnConfDir).listFiles((dir1, name) -> {
-                    if (name.endsWith(".xml")) {
-                        return true;
-                    }
-                    return false;
-                });
-                if (xmlFileList != null) {
-                    for (File xmlFile : xmlFileList) {
-                        yarnConf.addResource(xmlFile.toURI().toURL());
-                    }
-                }
-            }
-
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
+        FileUtils.listFilesInDirectory(new File(yarnConfDir).toPath(), this::isXmlFile)
+                .stream()
+                .map(FunctionUtils.uncheckedFunction(FileUtils::toURL))
+                .forEach(yarnConf::addResource);
 
         haYarnConf(yarnConf);
         return yarnConf;
@@ -101,8 +88,7 @@ private YarnConfiguration getYarnConf(String yarnConfDir) {
      */
     private org.apache.hadoop.conf.Configuration haYarnConf(org.apache.hadoop.conf.Configuration yarnConf) {
         Iterator<Map.Entry<String, String>> iterator = yarnConf.iterator();
-        while (iterator.hasNext()) {
-            Map.Entry<String, String> entry = iterator.next();
+        iterator.forEachRemaining((Map.Entry<String, String> entry) -> {
             String key = entry.getKey();
             String value = entry.getValue();
             if (key.startsWith("yarn.resourcemanager.hostname.")) {
@@ -112,7 +98,12 @@ private org.apache.hadoop.conf.Configuration haYarnConf(org.apache.hadoop.conf.C
                     yarnConf.set(addressKey, value + ":" + YarnConfiguration.DEFAULT_RM_PORT);
                 }
             }
-        }
+        });
+
         return yarnConf;
     }
+
+    private boolean isXmlFile(java.nio.file.Path file) {
+        return XML_FILE_EXTENSION.equals(org.apache.flink.shaded.guava18.com.google.common.io.Files.getFileExtension(file.toString()));
+    }
 }

launcher/src/main/java/com/dtstack/flink/sql/launcher/utils/JobGraphBuildUtil.java

@@ -22,11 +22,9 @@
 import com.dtstack.flink.sql.launcher.entity.JobParamsInfo;
 import com.dtstack.flink.sql.util.MathUtil;
 import com.dtstack.flink.sql.util.PluginUtil;
-import com.google.common.collect.Lists;
 import org.apache.commons.io.Charsets;
 import org.apache.commons.lang.BooleanUtils;
 import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.flink.api.common.cache.DistributedCache;
 import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.client.program.PackagedProgramUtils;
@@ -35,18 +33,15 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.util.function.FunctionUtils;
 
 import java.io.File;
 import java.io.UnsupportedEncodingException;
-import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLDecoder;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Properties;
 import java.util.stream.Collectors;
 
@@ -117,26 +112,21 @@ public static void fillJobGraphClassPath(JobGraph jobGraph) {
         Map<String, DistributedCache.DistributedCacheEntry> jobCacheFileConfig = jobGraph.getUserArtifacts();
         List<URL> classPath = jobCacheFileConfig.entrySet().stream()
                 .filter(tmp -> tmp.getKey().startsWith("class_path"))
-                .map(tmp -> {
-                    try {
-                        return new URL("file:" + tmp.getValue().filePath);
-                    } catch (MalformedURLException e) {
-                        throw new RuntimeException(ExceptionUtils.getFullStackTrace(e));
-                    }
-                })
+                .map(FunctionUtils.uncheckedFunction(tmp -> new URL("file:" + tmp.getValue().filePath)))
                 .collect(Collectors.toList());
+
         jobGraph.getUserArtifacts().clear();
         jobGraph.setClasspaths(classPath);
     }
 
     public static List<File> getPluginPathToShipFiles(JobGraph jobGraph) {
-        List<File> shipFiles = new ArrayList<>();
-        Map<String, DistributedCache.DistributedCacheEntry> jobCacheFileConfig = jobGraph.getUserArtifacts();
-        for (Map.Entry<String, DistributedCache.DistributedCacheEntry> tmp : jobCacheFileConfig.entrySet()) {
-            if (tmp.getKey().startsWith("class_path")) {
-                shipFiles.add(new File(tmp.getValue().filePath));
-            }
-        }
+       List<File> shipFiles = jobGraph.getUserArtifacts()
+                .entrySet()
+                .stream()
+                .filter(tmp -> tmp.getKey().startsWith("class_path"))
+                .map(tmp -> new File(tmp.getValue().filePath))
+                .collect(Collectors.toList());
+
         jobGraph.getUserArtifacts().clear();
         return shipFiles;
     }