异步维表

Author: dapeng

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

@@ -98,18 +98,26 @@ public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         AbstractSideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
         HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
+        Map<String, Object> hbaseConfig = hbaseSideTableInfo.getHbaseConfig();
 
         ExecutorService executorService =new ThreadPoolExecutor(DEFAULT_POOL_SIZE, DEFAULT_POOL_SIZE,
                 0L, TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<>(), new DTThreadFactory("hbase-aysnc"));
 
         Config config = new Config();
         config.overrideConfig(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_QUORUM, hbaseSideTableInfo.getHost());
-        config.overrideConfig(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM_ASYNC, hbaseSideTableInfo.getParent());
-
-        if (hbaseSideTableInfo.isKerberosAuthEnable()) {
-            fillAsyncKerberosConfig(config, hbaseSideTableInfo);
+        config.overrideConfig(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM, hbaseSideTableInfo.getParent());
+        HbaseConfigUtils.loadKrb5Conf(hbaseConfig);
+        hbaseConfig.entrySet().forEach(entity -> {
+            config.overrideConfig(entity.getKey(), (String) entity.getValue());
+        });
+
+        if (HbaseConfigUtils.asyncOpenKerberos(hbaseConfig)) {
+            String jaasStr = HbaseConfigUtils.buildJaasStr(hbaseConfig);
+            String jaasFilePath = HbaseConfigUtils.creatJassFile(jaasStr);
+            config.overrideConfig(HbaseConfigUtils.KEY_JAVA_SECURITY_AUTH_LOGIN_CONF, jaasFilePath);
         }
+
         hBaseClient = new HBaseClient(config, executorService);
 
         try {