增加缺失的参数

dapeng · dapeng · commit 5d71d4a4fcd8 · 2020-05-27T14:07:20.000+08:00
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseConfigUtils.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseConfigUtils.java
@@ -18,19 +18,12 @@
 
 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;
 
 /**
  *
@@ -52,13 +45,18 @@ public class HbaseConfigUtils {
     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 final static String KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM = "zookeeper.znode.parent";
+
+    public final static String KEY_HBASE_CLIENT_KEYTAB_FILE = "hbase.client.keytab.file";
+    public final static String KEY_HBASE_CLIENT_KERBEROS_PRINCIPAL = "hbase.client.kerberos.principal";
 
 
     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 {
+        LOG.info("loginAndReturnUGI principal {}",principal);
+        LOG.info("loginAndReturnUGI keytab {}",keytab);
         if (conf == null) {
             throw new IllegalArgumentException("kerberos conf can not be null");
         }
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java
@@ -29,8 +29,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.*;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.client.Delete;
@@ -70,6 +69,8 @@ public class HbaseOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private String regionserverPrincipal;
     private String securityKrb5Conf;
     private String zookeeperSaslClient;
+    private String clientPrincipal;
+    private String clientKeytabFile;
 
     private String[] families;
     private String[] qualifiers;
@@ -78,48 +79,78 @@ public class HbaseOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private transient Connection conn;
     private transient Table table;
 
+    private transient ChoreService choreService;
+
     @Override
     public void configure(Configuration parameters) {
         LOG.warn("---configure---");
         conf = HBaseConfiguration.create();
-        conf.set("hbase.zookeeper.quorum", host);
-        if (zkParent != null && !"".equals(zkParent)) {
-            conf.set("zookeeper.znode.parent", zkParent);
-        }
-        LOG.warn("---configure end ---");
     }
 
     @Override
     public void open(int taskNumber, int numTasks) throws IOException {
         LOG.warn("---open---");
-        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);
-        }
+        openConn();
         table = conn.getTable(TableName.valueOf(tableName));
         LOG.warn("---open end(get table from hbase) ---");
         initMetric();
     }
 
+    private void openConn(){
+        try{
+            if (kerberosAuthEnable) {
+                LOG.info("open kerberos conn");
+                openKerberosConn();
+            } else {
+                LOG.info("open conn");
+                conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_QUORUM, host);
+                conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM, zkParent);
+                conn = ConnectionFactory.createConnection(conf);
+            }
+        }catch (Exception e){
+            throw new RuntimeException(e);
+        }
+
+    }
+    private void openKerberosConn() throws IOException {
+        conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_QUORUM, host);
+        conf.set(HbaseConfigUtils.KEY_HBASE_ZOOKEEPER_ZNODE_QUORUM, zkParent);
+
+        LOG.info("kerberos config:{}", this.toString());
+        Preconditions.checkArgument(!StringUtils.isEmpty(clientPrincipal), " clientPrincipal not null!");
+        Preconditions.checkArgument(!StringUtils.isEmpty(clientKeytabFile), " clientKeytabFile not null!");
+
+        fillSyncKerberosConfig(conf, regionserverPrincipal, zookeeperSaslClient, securityKrb5Conf);
+
+        clientKeytabFile = System.getProperty("user.dir") + File.separator + clientKeytabFile;
+        clientPrincipal = !StringUtils.isEmpty(clientPrincipal) ? clientPrincipal : regionserverPrincipal;
+
+        conf.set(HbaseConfigUtils.KEY_HBASE_CLIENT_KEYTAB_FILE, clientKeytabFile);
+        conf.set(HbaseConfigUtils.KEY_HBASE_CLIENT_KERBEROS_PRINCIPAL, clientPrincipal);
+
+        UserGroupInformation userGroupInformation = HbaseConfigUtils.loginAndReturnUGI(conf, clientPrincipal, clientKeytabFile);
+        org.apache.hadoop.conf.Configuration finalConf = conf;
+        conn = userGroupInformation.doAs(new PrivilegedAction<Connection>() {
+            @Override
+            public Connection run() {
+                try {
+                    ScheduledChore authChore = AuthUtil.getAuthChore(finalConf);
+                    if (authChore != null) {
+                        choreService = new ChoreService("hbaseKerberosSink");
+                        choreService.scheduleChore(authChore);
+                    }
+
+                    return ConnectionFactory.createConnection(finalConf);
+                } catch (IOException e) {
+                    LOG.error("Get connection fail with config:{}", finalConf);
+                    throw new RuntimeException(e);
+                }
+            }
+        });
+    }
+
+
+
     @Override
     public void writeRecord(Tuple2 tuple2) {
         Tuple2<Boolean, Row> tupleTrans = tuple2;
@@ -227,7 +258,6 @@ public void close() throws IOException {
             conn = null;
         }
     }
-
     private HbaseOutputFormat() {
     }
 
@@ -264,11 +294,6 @@ public HbaseOutputFormatBuilder setRowkey(String rowkey) {
             return this;
         }
 
-        public HbaseOutputFormatBuilder setUpdateMode(String updateMode) {
-            format.updateMode = updateMode;
-            return this;
-        }
-
         public HbaseOutputFormatBuilder setColumnNames(String[] columnNames) {
             format.columnNames = columnNames;
             return this;
@@ -283,6 +308,7 @@ public HbaseOutputFormatBuilder setColumnNameFamily(Map<String, String> columnNa
             format.columnNameFamily = columnNameFamily;
             return this;
         }
+
         public HbaseOutputFormatBuilder setKerberosAuthEnable(boolean kerberosAuthEnable) {
             format.kerberosAuthEnable = kerberosAuthEnable;
             return this;
@@ -308,6 +334,16 @@ public HbaseOutputFormatBuilder setZookeeperSaslClient(String zookeeperSaslClien
             return this;
         }
 
+        public HbaseOutputFormatBuilder setClientPrincipal(String clientPrincipal) {
+            format.clientPrincipal = clientPrincipal;
+            return this;
+        }
+
+        public HbaseOutputFormatBuilder setClientKeytabFile(String clientKeytabFile) {
+            format.clientKeytabFile = clientKeytabFile;
+            return this;
+        }
+
 
         public HbaseOutputFormat finish() {
             Preconditions.checkNotNull(format.host, "zookeeperQuorum should be specified");
@@ -323,7 +359,7 @@ public HbaseOutputFormat finish() {
                 String[] columns = keySet.toArray(new String[keySet.size()]);
                 for (int i = 0; i < columns.length; ++i) {
                     String col = columns[i];
-                    String[] part = StringUtils.split(col, ":");;
+                    String[] part = col.split(":");
                     families[i] = part[0];
                     qualifiers[i] = part[1];
                 }
@@ -336,16 +372,8 @@ public HbaseOutputFormat finish() {
 
     }
 
-    private void fillSyncKerberosConfig( org.apache.hadoop.conf.Configuration config, String regionserverKeytabFile, String regionserverPrincipal,
+    private void fillSyncKerberosConfig(org.apache.hadoop.conf.Configuration config, 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");
         }
@@ -366,5 +394,17 @@ private void fillSyncKerberosConfig( org.apache.hadoop.conf.Configuration config
         }
     }
 
+    @Override
+    public String toString() {
+        return "HbaseOutputFormat kerberos{" +
+                "kerberosAuthEnable=" + kerberosAuthEnable +
+                ", regionserverKeytabFile='" + regionserverKeytabFile + '\'' +
+                ", regionserverPrincipal='" + regionserverPrincipal + '\'' +
+                ", securityKrb5Conf='" + securityKrb5Conf + '\'' +
+                ", zookeeperSaslClient='" + zookeeperSaslClient + '\'' +
+                ", clientPrincipal='" + clientPrincipal + '\'' +
+                ", clientKeytabFile='" + clientKeytabFile + '\'' +
+                '}';
+    }
 
 }
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseSink.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseSink.java
@@ -59,6 +59,9 @@ public class HbaseSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
     protected String regionserverPrincipal;
     protected String securityKrb5Conf;
     protected String zookeeperSaslClient;
+
+    private String clientPrincipal;
+    private String clientKeytabFile;
     private int parallelism = -1;
 
 
@@ -83,6 +86,9 @@ public HbaseSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.securityKrb5Conf = hbaseTableInfo.getSecurityKrb5Conf();
         this.zookeeperSaslClient = hbaseTableInfo.getZookeeperSaslClient();
 
+        this.clientKeytabFile = hbaseTableInfo.getClientKeytabFile();
+        this.clientPrincipal = hbaseTableInfo.getClientPrincipal();
+
         Integer tmpSinkParallelism = hbaseTableInfo.getParallelism();
         if (tmpSinkParallelism != null) {
             this.parallelism = tmpSinkParallelism;
@@ -104,6 +110,9 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         builder.setSecurityKrb5Conf(securityKrb5Conf);
         builder.setZookeeperSaslClient(zookeeperSaslClient);
 
+        builder.setClientPrincipal(clientPrincipal);
+        builder.setClientKeytabFile(clientKeytabFile);
+
         HbaseOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);
         dataStream.addSink(richSinkFunction);
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseSinkParser.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseSinkParser.java
@@ -61,6 +61,10 @@ public class HbaseSinkParser extends AbstractTableParser {
     public static final String SECURITY_KRB5_CONF_KEY = "securityKrb5Conf";
     public static final String ZOOKEEPER_SASL_CLINT_KEY = "zookeeperSaslClient";
 
+    public static final String CLIENT_PRINCIPAL_KEY = "clientPrincipal";
+    public static final String CLIENT_KEYTABFILE_KEY = "clientKeytabFile";
+
+
     @Override
     protected boolean fieldNameNeedsUpperCase() {
         return false;
@@ -85,6 +89,9 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         hbaseTableInfo.setRegionserverPrincipal((String) props.get(REGIONSERVER_PRINCIPAL_KEY.toLowerCase()));
         hbaseTableInfo.setSecurityKrb5Conf((String) props.get(SECURITY_KRB5_CONF_KEY.toLowerCase()));
         hbaseTableInfo.setZookeeperSaslClient((String) props.get(ZOOKEEPER_SASL_CLINT_KEY.toLowerCase()));
+
+        hbaseTableInfo.setClientPrincipal((String) props.get(CLIENT_PRINCIPAL_KEY.toLowerCase()));
+        hbaseTableInfo.setClientKeytabFile((String) props.get(CLIENT_KEYTABFILE_KEY.toLowerCase()));
         return hbaseTableInfo;
     }
 
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseTableInfo.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseTableInfo.java
@@ -66,6 +66,10 @@ public class HbaseTableInfo extends AbstractTargetTableInfo {
 
     private String zookeeperSaslClient;
 
+    private String clientPrincipal;
+
+    private String clientKeytabFile;
+
     private Map<String,Object> hbaseConfig = Maps.newHashMap();
 
     public HbaseTableInfo(){
@@ -212,4 +216,20 @@ public void setZookeeperSaslClient(String zookeeperSaslClient) {
         this.zookeeperSaslClient = zookeeperSaslClient;
     }
 
+    public String getClientPrincipal() {
+        return clientPrincipal;
+    }
+
+    public void setClientPrincipal(String clientPrincipal) {
+        this.clientPrincipal = clientPrincipal;
+    }
+
+    public String getClientKeytabFile() {
+        return clientKeytabFile;
+    }
+
+    public void setClientKeytabFile(String clientKeytabFile) {
+        this.clientKeytabFile = clientKeytabFile;
+    }
+
 }
