Kerberos认证的hbase 维表修复

Author: dapeng

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncReqRow.java

@@ -20,20 +20,16 @@
 
 package com.dtstack.flink.sql.side.hbase;
 
-import com.dtstack.flink.sql.enums.ECacheContentType;
 import com.dtstack.flink.sql.side.BaseAsyncReqRow;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
-import com.dtstack.flink.sql.side.cache.CacheObj;
 import com.dtstack.flink.sql.side.hbase.rowkeydealer.AbstractRowKeyModeDealer;
 import com.dtstack.flink.sql.side.hbase.rowkeydealer.PreRowKeyModeDealerDealer;
 import com.dtstack.flink.sql.side.hbase.rowkeydealer.RowKeyEqualModeDealer;
 import com.dtstack.flink.sql.side.hbase.table.HbaseSideTableInfo;
 import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.hbase.utils.HbaseConfigUtils;
-import com.dtstack.flink.sql.util.AuthUtil;
-import com.google.common.collect.Maps;
 import com.stumbleupon.async.Deferred;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -42,15 +38,14 @@
 import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
+import org.apache.hadoop.security.authentication.util.KerberosName;
 import org.hbase.async.Config;
 import org.hbase.async.HBaseClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import sun.security.krb5.KrbException;
 
-import java.io.File;
-import java.io.IOException;
 import java.sql.Timestamp;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.TimeZone;
@@ -118,9 +113,10 @@ public void open(Configuration parameters) throws Exception {
         if (HbaseConfigUtils.asyncOpenKerberos(hbaseConfig)) {
             String jaasStr = HbaseConfigUtils.buildJaasStr(hbaseConfig);
             String jaasFilePath = HbaseConfigUtils.creatJassFile(jaasStr);
+            System.setProperty(HbaseConfigUtils.KEY_JAVA_SECURITY_AUTH_LOGIN_CONF, jaasFilePath);
             config.overrideConfig(HbaseConfigUtils.KEY_JAVA_SECURITY_AUTH_LOGIN_CONF, jaasFilePath);
+            refreshConfig();
         }
-
         hBaseClient = new HBaseClient(config, executorService);
 
         try {
@@ -148,6 +144,13 @@ public void open(Configuration parameters) throws Exception {
         }
     }
 
+    private void refreshConfig() throws KrbException {
+        sun.security.krb5.Config.refresh();
+        KerberosName.resetDefaultRealm();
+        //reload java.security.auth.login.config
+        javax.security.auth.login.Configuration.setConfiguration(null);
+    }
+
     @Override
     public void handleAsyncInvoke(Map<String, Object> inputParams, CRow input, ResultFuture<CRow> resultFuture) throws Exception {
         rowKeyMode.asyncGetData(tableName, buildCacheKey(inputParams), input, resultFuture, sideInfo.getSideCache());

hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/utils/HbaseConfigUtils.java

@@ -59,7 +59,7 @@ public class HbaseConfigUtils {
     private final static String KEY_HBASE_SECURITY_AUTH_ENABLE = "hbase.security.auth.enable";
     private final static String KEY_HBASE_SASL_CLIENTCONFIG = "hbase.sasl.clientconfig";
     private final static String KEY_HBASE_KERBEROS_REGIONSERVER_PRINCIPAL = "hbase.kerberos.regionserver.principal";
-    private static final String KEY_KEY_TAB = "hbase.keytab";
+    public static final String KEY_KEY_TAB = "hbase.keytab";
     private static final String KEY_PRINCIPAL = "hbase.principal";
 
     public final static String KEY_HBASE_ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";
@@ -159,7 +159,7 @@ public static String getKeytab(Map<String, Object> hbaseConfigMap) {
     }
 
     public static void loadKrb5Conf(Map<String, Object> kerberosConfig) {
-        String krb5FilePath = MapUtils.getString(kerberosConfig, KEY_JAVA_SECURITY_KRB5_CONF);
+        String krb5FilePath = System.getProperty("user.dir") + File.separator + MapUtils.getString(kerberosConfig, KEY_JAVA_SECURITY_KRB5_CONF);
         if (!org.apache.commons.lang.StringUtils.isEmpty(krb5FilePath)) {
             System.setProperty(KEY_JAVA_SECURITY_KRB5_CONF, krb5FilePath);;
         }
@@ -183,7 +183,7 @@ public static String buildJaasStr(Map<String, Object> kerberosConfig) {
             }
         }
 
-        String keyTab = MapUtils.getString(kerberosConfig, KEY_KEY_TAB);
+        String keyTab = System.getProperty("user.dir") + File.separator + MapUtils.getString(kerberosConfig, KEY_KEY_TAB);
         String principal = MapUtils.getString(kerberosConfig, KEY_PRINCIPAL);
 
         StringBuilder jaasSB = new StringBuilder("Client {\n" +