fix 补充参数

Author: dapeng

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseConfigUtils.java

@@ -0,0 +1,79 @@
+/*
+ * 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.sink.hbase;
+
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ *
+ *  The utility class of HBase connection
+ *
+ * Date: 2019/12/24
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class HbaseConfigUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(HbaseConfigUtils.class);
+    // sync side kerberos
+    public final static String KEY_HBASE_SECURITY_AUTHENTICATION = "hbase.security.authentication";
+    public final static String KEY_HBASE_SECURITY_AUTHORIZATION = "hbase.security.authorization";
+    public final static String KEY_HBASE_MASTER_KEYTAB_FILE = "hbase.master.keytab.file";
+    public final static String KEY_HBASE_MASTER_KERBEROS_PRINCIPAL = "hbase.master.kerberos.principal";
+    public final static String KEY_HBASE_REGIONSERVER_KEYTAB_FILE = "hbase.regionserver.keytab.file";
+    public final static String KEY_HBASE_REGIONSERVER_KERBEROS_PRINCIPAL = "hbase.regionserver.kerberos.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";
+
+
+    public static final String KEY_JAVA_SECURITY_KRB5_CONF = "java.security.krb5.conf";
+    public static final String KEY_ZOOKEEPER_SASL_CLIENT = "zookeeper.sasl.client";
+
+    public static UserGroupInformation loginAndReturnUGI(Configuration conf, String principal, String keytab) throws IOException {
+        if (conf == null) {
+            throw new IllegalArgumentException("kerberos conf can not be null");
+        }
+
+        if (org.apache.commons.lang.StringUtils.isEmpty(principal)) {
+            throw new IllegalArgumentException("principal can not be null");
+        }
+
+        if (org.apache.commons.lang.StringUtils.isEmpty(keytab)) {
+            throw new IllegalArgumentException("keytab can not be null");
+        }
+
+        conf.set("hadoop.security.authentication", "Kerberos");
+        UserGroupInformation.setConfiguration(conf);
+
+        return UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytab);
+    }
+}

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java

@@ -37,10 +37,13 @@
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
 import java.io.IOException;
+import java.security.PrivilegedAction;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -91,6 +94,29 @@ public void open(int taskNumber, int numTasks) throws IOException {
         LOG.warn("---open---");
         conn = ConnectionFactory.createConnection(conf);
         table = conn.getTable(TableName.valueOf(tableName));
+        if (kerberosAuthEnable) {
+            conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_QUORUM, host);
+            conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM, zkParent);
+            fillSyncKerberosConfig(conf, regionserverKeytabFile, regionserverPrincipal, zookeeperSaslClient, securityKrb5Conf);
+
+            UserGroupInformation userGroupInformation = HbaseConfigUtils.loginAndReturnUGI(conf, regionserverPrincipal, regionserverKeytabFile);
+            org.apache.hadoop.conf.Configuration finalConf = conf;
+            conn = userGroupInformation.doAs(new PrivilegedAction<Connection>() {
+                @Override
+                public Connection run() {
+                    try {
+                        return ConnectionFactory.createConnection(finalConf);
+                    } catch (IOException e) {
+                        LOG.error("Get connection fail with config:{}", finalConf);
+                        throw new RuntimeException(e);
+                    }
+                }
+            });
+        } else {
+            conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_QUORUM, host);
+            conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM, zkParent);
+            conn = ConnectionFactory.createConnection(conf);
+        }
         LOG.warn("---open end(get table from hbase) ---");
         initMetric();
     }
@@ -311,5 +337,35 @@ public HbaseOutputFormat finish() {
 
     }
 
+    private void fillSyncKerberosConfig( org.apache.hadoop.conf.Configuration config, String regionserverKeytabFile, String regionserverPrincipal,
+                                        String zookeeperSaslClient, String securityKrb5Conf) throws IOException {
+        if (StringUtils.isEmpty(regionserverKeytabFile)) {
+            throw new IllegalArgumentException("Must provide regionserverKeytabFile when authentication is Kerberos");
+        }
+        String regionserverKeytabFilePath = System.getProperty("user.dir") + File.separator + regionserverKeytabFile;
+        LOG.info("regionserverKeytabFilePath:{}",regionserverKeytabFilePath);
+        config.set(HbaseConfigUtils.KEY_HBASE_MASTER_KEYTAB_FILE, regionserverKeytabFilePath);
+        config.set(HbaseConfigUtils.KEY_HBASE_REGIONSERVER_KEYTAB_FILE, regionserverKeytabFilePath);
+
+        if (StringUtils.isEmpty(regionserverPrincipal)) {
+            throw new IllegalArgumentException("Must provide regionserverPrincipal when authentication is Kerberos");
+        }
+        config.set(HbaseConfigUtils.KEY_HBASE_MASTER_KERBEROS_PRINCIPAL, regionserverPrincipal);
+        config.set(HbaseConfigUtils.KEY_HBASE_REGIONSERVER_KERBEROS_PRINCIPAL, regionserverPrincipal);
+        config.set(HbaseConfigUtils.KEY_HBASE_SECURITY_AUTHORIZATION, "true");
+        config.set(HbaseConfigUtils.KEY_HBASE_SECURITY_AUTHENTICATION, "kerberos");
+
+
+        if (!StringUtils.isEmpty(zookeeperSaslClient)) {
+            System.setProperty(HbaseConfigUtils.KEY_ZOOKEEPER_SASL_CLIENT, zookeeperSaslClient);
+        }
+
+        if (!StringUtils.isEmpty(securityKrb5Conf)) {
+            String krb5ConfPath = System.getProperty("user.dir") + File.separator + securityKrb5Conf;
+            LOG.info("krb5ConfPath:{}", krb5ConfPath);
+            System.setProperty(HbaseConfigUtils.KEY_JAVA_SECURITY_KRB5_CONF, krb5ConfPath);
+        }
+    }
+
 
 }