添加对hadoop(非Kerberos) 写入 hbase（开启kerberos）兼容问题

Author: xuchao

core/pom.xml

@@ -108,6 +108,18 @@
             <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-hadoop2</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-shaded-hadoop2</artifactId>
+            <version>2.7.5-1.8.1</version>
         </dependency>
 
         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-statebackend-rocksdb -->

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseConfigUtils.java

@@ -19,9 +19,13 @@
 package com.dtstack.flink.sql.sink.hbase;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.HadoopKerberosName;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authentication.util.KerberosName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import sun.security.krb5.Config;
+import sun.security.krb5.KrbException;
 
 import java.io.IOException;
 
@@ -54,7 +58,11 @@ public class HbaseConfigUtils {
     public static final String KEY_JAVA_SECURITY_KRB5_CONF = "java.security.krb5.conf";
     public static final String KEY_ZOOKEEPER_SASL_CLIENT = "zookeeper.sasl.client";
 
-    public static UserGroupInformation loginAndReturnUGI(Configuration conf, String principal, String keytab) throws IOException {
+    public static final String KEY_HADOOP_SECURITY_AUTHENTICATION = "hadoop.security.authentication";
+    public static final String KEY_HADOOP_SECURITY_AUTH_TO_LOCAL = "hadoop.security.auth_to_local";
+    public static final String KEY_HADOOP_SECURITY_AUTHORIZATION = "hadoop.security.authorization";
+
+    public static UserGroupInformation loginAndReturnUGI(Configuration conf, String principal, String keytab) throws IOException, KrbException {
         LOG.info("loginAndReturnUGI principal {}",principal);
         LOG.info("loginAndReturnUGI keytab {}",keytab);
         if (conf == null) {
@@ -69,7 +77,13 @@ public static UserGroupInformation loginAndReturnUGI(Configuration conf, String
             throw new IllegalArgumentException("keytab can not be null");
         }
 
-        conf.set("hadoop.security.authentication", "Kerberos");
+        conf.set(KEY_HADOOP_SECURITY_AUTHENTICATION, "Kerberos");
+        //conf.set("hadoop.security.auth_to_local", "DEFAULT");
+        conf.set(KEY_HADOOP_SECURITY_AUTH_TO_LOCAL, "RULE:[1:$1] RULE:[2:$1]");
+        conf.set(KEY_HADOOP_SECURITY_AUTHORIZATION, "true");
+
+        Config.refresh();
+        KerberosName.resetDefaultRealm();
         UserGroupInformation.setConfiguration(conf);
 
         return UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytab);

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java

@@ -112,7 +112,7 @@ private void openConn(){
         }
 
     }
-    private void openKerberosConn() throws IOException {
+    private void openKerberosConn() throws Exception {
         conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_QUORUM, host);
         conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM, zkParent);