Merge branch 'hotfix_1.10_4.0.x_30358' into '1.10_release_4.0.x'

Hotfix 1.10 4.0.x 30358

See merge request dt-insight-engine/flinkStreamSQL!114

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/util/DtFileUtils.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.util;
+
+import org.apache.flink.util.Preconditions;
+
+import java.io.File;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/21
+ **/
+public class DtFileUtils {
+    public static void checkExists(String path) {
+        File file = new File(path);
+        String errorMsg = "%s file is not exist!";
+        Preconditions.checkState(file.exists(), errorMsg, path);
+    }
+}

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

@@ -20,30 +20,36 @@
 
 package com.dtstack.flink.sql.side.hbase;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 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.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.DtFileUtils;
 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.runtime.security.DynamicConfiguration;
+import org.apache.flink.runtime.security.KerberosUtils;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.table.dataformat.BaseRow;
+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 javax.security.auth.login.AppConfigurationEntry;
+import java.io.File;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
@@ -95,21 +101,30 @@ public void open(Configuration parameters) throws Exception {
 
         ExecutorService executorService =new ThreadPoolExecutor(DEFAULT_POOL_SIZE, DEFAULT_POOL_SIZE,
                 0L, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(), new DTThreadFactory("hbase-aysnc"));
+                new LinkedBlockingQueue<>(), new DTThreadFactory("hbase-async"));
 
         Config config = new Config();
         config.overrideConfig(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_QUORUM, hbaseSideTableInfo.getHost());
         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);
-            System.setProperty(HbaseConfigUtils.KEY_JAVA_SECURITY_AUTH_LOGIN_CONF, jaasFilePath);
-            config.overrideConfig(HbaseConfigUtils.KEY_JAVA_SECURITY_AUTH_LOGIN_CONF, jaasFilePath);
+        if (HbaseConfigUtils.isEnableKerberos(hbaseConfig)) {
+            HbaseConfigUtils.loadKrb5Conf(hbaseConfig);
+            String principal = MapUtils.getString(hbaseConfig, HbaseConfigUtils.KEY_PRINCIPAL);
+            HbaseConfigUtils.checkOpt(principal, HbaseConfigUtils.KEY_PRINCIPAL);
+            String regionserverPrincipal = MapUtils.getString(hbaseConfig, HbaseConfigUtils.KEY_HBASE_KERBEROS_REGIONSERVER_PRINCIPAL);
+            HbaseConfigUtils.checkOpt(regionserverPrincipal, HbaseConfigUtils.KEY_HBASE_KERBEROS_REGIONSERVER_PRINCIPAL);
+            String keytab = MapUtils.getString(hbaseConfig, HbaseConfigUtils.KEY_KEY_TAB);
+            HbaseConfigUtils.checkOpt(keytab, HbaseConfigUtils.KEY_KEY_TAB);
+            String keytabPath = System.getProperty("user.dir") + File.separator + keytab;
+            DtFileUtils.checkExists(keytabPath);
+
+            LOG.info("Kerberos login with keytab: {} and principal: {}", keytab, principal);
+            String name = "HBaseClient";
+            config.overrideConfig("hbase.sasl.clientconfig", name);
+            appendJaasConf(name, keytab, principal);
             refreshConfig();
         }
 
@@ -144,7 +159,17 @@ 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);
+    }
+
+    private void appendJaasConf(String name, String keytab, String principal) {
+        javax.security.auth.login.Configuration priorConfig = javax.security.auth.login.Configuration.getConfiguration();
+        // construct a dynamic JAAS configuration
+        DynamicConfiguration currentConfig = new DynamicConfiguration(priorConfig);
+        // wire up the configured JAAS login contexts to use the krb5 entries
+        AppConfigurationEntry krb5Entry = KerberosUtils.keytabEntry(keytab, principal);
+        currentConfig.addAppConfigurationEntry(name, krb5Entry);
+        javax.security.auth.login.Configuration.setConfiguration(currentConfig);
     }
 
     @Override

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

@@ -18,6 +18,9 @@
 
 package com.dtstack.flink.sql.side.hbase.utils;
 
+import com.dtstack.flink.sql.util.DtFileUtils;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -26,9 +29,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.BufferedWriter;
 import java.io.File;
-import java.io.FileWriter;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
@@ -57,21 +58,15 @@ public class HbaseConfigUtils {
     // async side kerberos
     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";
+    public 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";
 
 
     private static final String KEY_JAVA_SECURITY_KRB5_CONF = "java.security.krb5.conf";
-    public static final String KEY_JAVA_SECURITY_AUTH_LOGIN_CONF = "java.security.auth.login.config";
-
-
-    private static final String SP = File.separator;
-    private static final  String KEY_KRB5_CONF = "krb5.conf";
-
 
     private static List<String> KEYS_KERBEROS_REQUIRED = Arrays.asList(
             KEY_HBASE_SECURITY_AUTHENTICATION,
@@ -100,23 +95,29 @@ public static Configuration getConfig(Map<String, Object> hbaseConfigMap) {
         return hConfiguration;
     }
 
-    public static boolean openKerberos(Map<String, Object> hbaseConfigMap) {
-        if (!MapUtils.getBooleanValue(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHORIZATION)) {
-            return false;
+    public static boolean isEnableKerberos(Map<String, Object> hbaseConfigMap) {
+        boolean hasAuthorization = AUTHENTICATION_TYPE.equalsIgnoreCase(
+                MapUtils.getString(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHORIZATION)
+        );
+        boolean hasAuthentication =  AUTHENTICATION_TYPE.equalsIgnoreCase(
+                MapUtils.getString(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHENTICATION)
+        );
+        boolean hasAuthEnable = MapUtils.getBooleanValue(hbaseConfigMap, KEY_HBASE_SECURITY_AUTH_ENABLE);
+
+        if(hasAuthentication || hasAuthorization || hasAuthEnable) {
+            LOG.info("Enable kerberos for hbase.");
+            setKerberosConf(hbaseConfigMap);
+            return true;
         }
-        return AUTHENTICATION_TYPE.equalsIgnoreCase(MapUtils.getString(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHENTICATION));
+        return false;
     }
 
-    public static boolean asyncOpenKerberos(Map<String, Object> hbaseConfigMap) {
-        if (!MapUtils.getBooleanValue(hbaseConfigMap, KEY_HBASE_SECURITY_AUTH_ENABLE)) {
-            return false;
-        }
-        return AUTHENTICATION_TYPE.equalsIgnoreCase(MapUtils.getString(hbaseConfigMap, KEY_HBASE_SECURITY_AUTHENTICATION));
+    private 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 Configuration getHadoopConfiguration(Map<String, Object> hbaseConfigMap) {
         for (String key : KEYS_KERBEROS_REQUIRED) {
             if (StringUtils.isEmpty(MapUtils.getString(hbaseConfigMap, key))) {
@@ -157,46 +158,20 @@ public static String getKeytab(Map<String, Object> hbaseConfigMap) {
         throw new IllegalArgumentException("");
     }
 
-    public static void loadKrb5Conf(Map<String, Object> kerberosConfig) {
-        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);;
-        }
-    }
-
-    public static String creatJassFile(String configStr) throws IOException {
-        String fileName = System.getProperty("user.dir");
-        File krbConf = new File(fileName);
-        File temp = File.createTempFile("JAAS", ".conf", krbConf);
-        temp.deleteOnExit();
-        BufferedWriter out = new BufferedWriter(new FileWriter(temp, false));
-        out.write(configStr + "\n");
-        out.close();
-        return temp.getAbsolutePath();
+    public static void loadKrb5Conf(Map<String, Object> config) {
+        String krb5conf = MapUtils.getString(config, KEY_JAVA_SECURITY_KRB5_CONF);
+        checkOpt(krb5conf, KEY_JAVA_SECURITY_KRB5_CONF);
+        String krb5FilePath = System.getProperty("user.dir") + File.separator + MapUtils.getString(config, KEY_JAVA_SECURITY_KRB5_CONF);
+        DtFileUtils.checkExists(krb5FilePath);
+        System.setProperty(KEY_JAVA_SECURITY_KRB5_CONF, krb5FilePath);
+        LOG.info("{} is set to {}", KEY_JAVA_SECURITY_KRB5_CONF, krb5FilePath);
     }
 
-    public static String buildJaasStr(Map<String, Object> kerberosConfig) {
-        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" +
-                "  useTicketCache=false\n");
-        jaasSB.append(" keyTab=\"").append(keyTab).append("\"").append("\n");
-        jaasSB.append(" principal=\"").append(principal).append("\"").append(";\n");
-        jaasSB.append("};");
-        return jaasSB.toString();
+    // TODO 日后改造可以下沉到Core模块
+    public static void checkOpt(String opt, String key) {
+        Preconditions.checkState(!Strings.isNullOrEmpty(opt), "%s must be set!", key);
     }
 
-
-
     public static UserGroupInformation loginAndReturnUGI(Configuration conf, String principal, String keytab) throws IOException {
         if (conf == null) {
             throw new IllegalArgumentException("kerberos conf can not be null");