fix bug 30358

Author: simenliuxing

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

@@ -31,19 +31,23 @@
 import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.hbase.utils.HbaseConfigUtils;
 import com.stumbleupon.async.Deferred;
+import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
 import org.apache.flink.table.dataformat.BaseRow;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.hadoop.security.UserGroupInformation;
 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.security.PrivilegedExceptionAction;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
@@ -105,15 +109,29 @@ public void open(Configuration parameters) throws Exception {
             config.overrideConfig(entity.getKey(), (String) entity.getValue());
         });
 
+        String principal = null;
+        String keyTab = null;
         if (HbaseConfigUtils.asyncOpenKerberos(hbaseConfig)) {
-            String jaasStr = HbaseConfigUtils.buildJaasStr(hbaseConfig);
+            principal = MapUtils.getString(hbaseConfig, HbaseConfigUtils.KEY_PRINCIPAL);
+            keyTab = System.getProperty("user.dir") + File.separator + MapUtils.getString(hbaseConfig, HbaseConfigUtils.KEY_KEY_TAB);
+            String jaasStr = HbaseConfigUtils.buildJaasStr(hbaseConfig, principal, keyTab);
             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);
+        hBaseClient = null;
+        if (HbaseConfigUtils.asyncOpenKerberos(hbaseConfig)) {
+            hBaseClient = UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keyTab).doAs(new PrivilegedExceptionAction<HBaseClient>() {
+                @Override
+                public HBaseClient run() throws Exception {
+                    return new HBaseClient(config, executorService);
+                }
+            });
+        } else {
+            hBaseClient = new HBaseClient(config, executorService);
+        }
 
         try {
             Deferred deferred = hBaseClient.ensureTableExists(tableName)
@@ -144,7 +162,7 @@ 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);
+        // javax.security.auth.login.Configuration.setConfiguration(null);
     }
 
     @Override

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_SASL_CLIENTCONFIG = "hbase.sasl.clientconfig";
     private final static String KEY_HBASE_KERBEROS_REGIONSERVER_PRINCIPAL = "hbase.kerberos.regionserver.principal";
     public static final String KEY_KEY_TAB = "hbase.keytab";
-    private static final String KEY_PRINCIPAL = "hbase.principal";
+    public static final String KEY_PRINCIPAL = "hbase.principal";
 
     public final static String KEY_HBASE_ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";
     public final static String KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM = "hbase.zookeeper.znode.parent";
@@ -175,16 +175,13 @@ public static String creatJassFile(String configStr) throws IOException {
         return temp.getAbsolutePath();
     }
 
-    public static String buildJaasStr(Map<String, Object> kerberosConfig) {
+    public static String buildJaasStr(Map<String, Object> kerberosConfig,String principal,String keyTab) {
         for (String key : ASYNC_KEYS_KERBEROS_REQUIRED) {
             if (StringUtils.isEmpty(MapUtils.getString(kerberosConfig, key))) {
                 throw new IllegalArgumentException(String.format("Must provide [%s] when authentication is Kerberos", key));
             }
         }
 
-        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" +
                 "  com.sun.security.auth.module.Krb5LoginModule required\n" +
                 "  useKeyTab=true\n" +