hotfix【30257】简化对kerberos条件判断，认证之前刷新keytab

Author: dujie

flinkx-core/src/main/java/com/dtstack/flinkx/authenticate/KerberosUtil.java

@@ -25,6 +25,7 @@
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 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;
@@ -33,6 +34,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.util.Map;
 
 /**
@@ -122,6 +124,7 @@ public static void loadKrb5Conf(Map<String, Object> kerberosConfig){
 
         krb5FilePath = loadFile(kerberosConfig, krb5FilePath);
         kerberosConfig.put(KEY_JAVA_SECURITY_KRB5_CONF, krb5FilePath);
+        System.setProperty(KEY_JAVA_SECURITY_KRB5_CONF, krb5FilePath);
     }
 
     /**
@@ -249,4 +252,21 @@ public static String getPrincipalFileName(Map<String, Object> config) {
 
         return fileName;
     }
+
+    /**
+     * 刷新krb内容信息
+     */
+    public static void refreshConfig() {
+        try{
+            sun.security.krb5.Config.refresh();
+            Field defaultRealmField = KerberosName.class.getDeclaredField("defaultRealm");
+            defaultRealmField.setAccessible(true);
+            defaultRealmField.set(null, org.apache.hadoop.security.authentication.util.KerberosUtil.getDefaultRealm());
+            //reload java.security.auth.login.config
+            javax.security.auth.login.Configuration.setConfiguration(null);
+        }catch (Exception e){
+            LOG.warn("resetting default realm failed, current default realm will still be used.", e);
+        }
+
+    }
 }

flinkx-hbase/flinkx-hbase-core/src/main/java/com/dtstack/flinkx/hbase/HbaseHelper.java

@@ -53,6 +53,7 @@ public class HbaseHelper {
     private final static String AUTHENTICATION_TYPE = "Kerberos";
     private final static String KEY_HBASE_SECURITY_AUTHENTICATION = "hbase.security.authentication";
     private final static String KEY_HBASE_SECURITY_AUTHORIZATION = "hbase.security.authorization";
+    private final static String KEY_HBASE_SECURITY_AUTH_ENABLE = "hbase.security.auth.enable";
 
     public static org.apache.hadoop.hbase.client.Connection getHbaseConnection(Map<String,Object> hbaseConfigMap) {
         Validate.isTrue(MapUtils.isNotEmpty(hbaseConfigMap), "hbaseConfig不能为空Map结构!");
@@ -72,6 +73,7 @@ public static org.apache.hadoop.hbase.client.Connection getHbaseConnection(Map<S
 
     private static org.apache.hadoop.hbase.client.Connection getConnectionWithKerberos(Map<String,Object> hbaseConfigMap){
         try {
+            setKerberosConf(hbaseConfigMap);
             UserGroupInformation ugi = getUgi(hbaseConfigMap);
             return ugi.doAs(new PrivilegedAction<Connection>() {
                 @Override
@@ -118,11 +120,25 @@ public static Configuration getConfig(Map<String,Object> hbaseConfigMap){
     }
 
     public static boolean openKerberos(Map<String,Object> hbaseConfigMap){
-        if(!MapUtils.getBooleanValue(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHORIZATION)){
-            return false;
+        if(AUTHENTICATION_TYPE.equalsIgnoreCase(MapUtils.getString(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHORIZATION))
+                || AUTHENTICATION_TYPE.equalsIgnoreCase(MapUtils.getString(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHENTICATION))
+                || MapUtils.getBooleanValue(hbaseConfigMap, KEY_HBASE_SECURITY_AUTH_ENABLE)){
+            LOG.info("open kerberos for hbase.");
+            return true;
         }
 
-        return AUTHENTICATION_TYPE.equalsIgnoreCase(MapUtils.getString(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHENTICATION));
+        return false;
+    }
+
+
+    /**
+     * 设置hbase 开启kerberos 连接必要的固定参数
+     * @param hbaseConfigMap
+     */
+    public static void setKerberosConf(Map<String,Object> hbaseConfigMap){
+        hbaseConfigMap.put(KEY_HBASE_SECURITY_AUTHORIZATION, AUTHENTICATION_TYPE);
+        hbaseConfigMap.put(KEY_HBASE_SECURITY_AUTHENTICATION, AUTHENTICATION_TYPE);
+        hbaseConfigMap.put(KEY_HBASE_SECURITY_AUTH_ENABLE, true);
     }
 
     public static RegionLocator getRegionLocator(Connection hConnection, String userTable){