统一参数基于flink-conf.yaml,

覆盖参数基于confProp

Author: zoudaokoulife

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobClusterClientBuilder.java

@@ -26,6 +26,8 @@
 import org.apache.flink.configuration.Configuration;
 import com.google.common.base.Strings;
 import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.security.SecurityConfiguration;
+import org.apache.flink.runtime.security.SecurityUtils;
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.hadoop.fs.Path;
@@ -55,45 +57,42 @@ public class PerJobClusterClientBuilder {
 
     private static final Logger LOG = LoggerFactory.getLogger(PerJobClusterClientBuilder.class);
 
-    private static String KEYTAB = "security.kerberos.login.keytab";
-
-    private static String PRINCIPAL = "security.kerberos.login.principal";
+    private static final String DEFAULT_CONF_DIR = "./";
 
     private YarnClient yarnClient;
 
     private YarnConfiguration yarnConf;
 
-    public void init(String yarnConfDir, Properties conf) throws IOException {
+    private Configuration flinkConfig;
+
+    public void init(String yarnConfDir, Configuration flinkConfig, Properties userConf) throws Exception {
 
         if(Strings.isNullOrEmpty(yarnConfDir)) {
             throw new RuntimeException("parameters of yarn is required");
         }
 
-        yarnConf = YarnConfLoader.getYarnConf(yarnConfDir);
+        userConf.forEach((key, val) -> flinkConfig.setString(key.toString(), val.toString()));
 
-        if (isKerberos(conf)){
-            String keytab = (String) conf.get(KEYTAB);
-            String principal = (String) conf.get(PRINCIPAL);
-            login(yarnConf, keytab, principal);
-        }
+        this.flinkConfig = flinkConfig;
+        SecurityUtils.install(new SecurityConfiguration(flinkConfig));
 
+        yarnConf = YarnConfLoader.getYarnConf(yarnConfDir);
         yarnClient = YarnClient.createYarnClient();
         yarnClient.init(yarnConf);
         yarnClient.start();
 
         System.out.println("----init yarn success ----");
     }
 
-    public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(Properties confProp, String flinkJarPath, Options launcherOptions, JobGraph jobGraph, Configuration flinkConfig)
+    public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(String flinkJarPath, Options launcherOptions, JobGraph jobGraph)
             throws MalformedURLException {
 
-        confProp.forEach((key, val) -> flinkConfig.setString(key.toString(), val.toString()));
-        String flinkConf = StringUtils.isEmpty(launcherOptions.getFlinkconf()) ? "" : launcherOptions.getFlinkconf();
+        String flinkConf = StringUtils.isEmpty(launcherOptions.getFlinkconf()) ? DEFAULT_CONF_DIR : launcherOptions.getFlinkconf();
         AbstractYarnClusterDescriptor clusterDescriptor = getClusterDescriptor(flinkConfig, yarnConf, flinkConf);
 
         if (StringUtils.isNotBlank(flinkJarPath)) {
             if (!new File(flinkJarPath).exists()) {
-                throw new RuntimeException("The Flink jar path is not exist");
+                throw new RuntimeException("The param '-flinkJarPath' ref dir is not exist");
             }
         }
 
@@ -163,22 +162,4 @@ private AbstractYarnClusterDescriptor getClusterDescriptor(
                 false);
     }
 
-    private boolean isKerberos(Properties conf){
-        String keytab = (String) conf.get(KEYTAB);
-        if (StringUtils.isNotBlank(keytab)){
-            return true;
-        } else {
-            return false;
-        }
-    }
-
-    private void login(org.apache.hadoop.conf.Configuration conf, String keytab, String principal) throws IOException {
-        if (StringUtils.isEmpty(principal)){
-            throw new RuntimeException(PRINCIPAL + " must not be null!");
-        }
-        UserGroupInformation.setConfiguration(conf);
-        UserGroupInformation.loginUserFromKeytab(principal, keytab);
-        LOG.info("login successfully! keytab: " + keytab + "principal: " + principal);
-        LOG.info("UGI: " + UserGroupInformation.getCurrentUser());
-    }
 }

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobSubmitter.java

@@ -62,11 +62,10 @@ public static String submit(Options launcherOptions, JobGraph jobGraph, Configur
         ClusterSpecification clusterSpecification = FLinkPerJobResourceUtil.createClusterSpecification(confProperties);
 
         PerJobClusterClientBuilder perJobClusterClientBuilder = new PerJobClusterClientBuilder();
-        perJobClusterClientBuilder.init(launcherOptions.getYarnconf(), confProperties);
+        perJobClusterClientBuilder.init(launcherOptions.getYarnconf(), flinkConfig, confProperties);
 
         String flinkJarPath = launcherOptions.getFlinkJarPath();
-
-        AbstractYarnClusterDescriptor yarnClusterDescriptor = perJobClusterClientBuilder.createPerJobClusterDescriptor(confProperties, flinkJarPath, launcherOptions, jobGraph, flinkConfig);
+        AbstractYarnClusterDescriptor yarnClusterDescriptor = perJobClusterClientBuilder.createPerJobClusterDescriptor(flinkJarPath, launcherOptions, jobGraph);
         ClusterClient<ApplicationId> clusterClient = yarnClusterDescriptor.deployJobCluster(clusterSpecification, jobGraph,true);
 
         String applicationId = clusterClient.getClusterId().toString();