Merge branch 'v1.9.0_dev_kerberos' into 'v1.9.0_dev'

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



See merge request !166

Author: zoudaokoulife

core/pom.xml

@@ -67,9 +67,9 @@
         </dependency>
 
         <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-yarn_2.11</artifactId>
-            <version>${flink.version}</version>
+            <groupId>commons-codec</groupId>
+            <artifactId>commons-codec</artifactId>
+            <version>1.10</version>
         </dependency>
 
         <dependency>

launcher/pom.xml

@@ -22,6 +22,42 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-yarn_2.11</artifactId>
             <version>${flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-hadoop-2</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-common</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-client</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-mapreduce-client-core</artifactId>
+            <version>${hadoop.version}</version>
         </dependency>
 
         <dependency>

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

@@ -29,10 +29,14 @@
 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 java.io.File;
+import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;
@@ -48,16 +52,31 @@
  */
 
 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 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);
+
+        if (isKerberos(conf)){
+            String keytab = (String) conf.get(KEYTAB);
+            String principal = (String) conf.get(PRINCIPAL);
+            login(yarnConf, keytab, principal);
+        }
+
         yarnClient = YarnClient.createYarnClient();
         yarnClient.init(yarnConf);
         yarnClient.start();
@@ -141,4 +160,23 @@ private AbstractYarnClusterDescriptor getClusterDescriptor(
                 yarnClient,
                 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

@@ -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();
 

pom.xml

@@ -36,6 +36,7 @@
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <flink.version>1.9.1</flink.version>
+      <hadoop.version>2.7.3</hadoop.version>
   </properties>
 
     <build>