[kerberos][在启动yarnclient之前添加认证信息]

Author: yanxi0227

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

@@ -29,10 +29,15 @@
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.rmi.runtime.Log;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;
@@ -48,16 +53,38 @@
  */
 
 public class PerJobClusterClientBuilder {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PerJobClusterClientBuilder.class);
+
+    private static String SECURITY = "hadoop.security.authorization";
+
+    private static String KEYTAB = "security.kerberos.login.keytab";
+
+    private static String PRINCIPAL = "security.kerberos.login.principal";
+
     private YarnClient yarnClient;
 
     private YarnConfiguration yarnConf;
 
-    public void init(String yarnConfDir){
+    public void init(String yarnConfDir, Properties conf) throws IOException {
+
         if(Strings.isNullOrEmpty(yarnConfDir)) {
             throw new RuntimeException("parameters of yarn is required");
         }
 
         yarnConf = YarnConfLoader.getYarnConf(yarnConfDir);
+
+        Boolean security = yarnConf.getBoolean(SECURITY, false);
+
+        String keytab = (String) conf.get(KEYTAB);
+        String principal = (String) conf.get(PRINCIPAL);
+        if (security && !Strings.isNullOrEmpty(keytab)){
+            UserGroupInformation.setConfiguration(yarnConf);
+            UserGroupInformation.loginUserFromKeytab(keytab, principal);
+            LOG.info("login successfully! keytab: " + keytab + "principal: " + principal);
+            LOG.info("UGI: " + UserGroupInformation.getCurrentUser());
+        }
+
         yarnClient = YarnClient.createYarnClient();
         yarnClient.init(yarnConf);
         yarnClient.start();

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

@@ -61,7 +61,7 @@ public static String submit(Options launcherOptions, JobGraph jobGraph) throws E
         ClusterSpecification clusterSpecification = FLinkPerJobResourceUtil.createClusterSpecification(confProperties);
 
         PerJobClusterClientBuilder perJobClusterClientBuilder = new PerJobClusterClientBuilder();
-        perJobClusterClientBuilder.init(launcherOptions.getYarnconf());
+        perJobClusterClientBuilder.init(launcherOptions.getYarnconf(), confProperties);
 
         String flinkJarPath = launcherOptions.getFlinkJarPath();