hbase kerberos sink

Author: dapeng

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

@@ -62,6 +62,12 @@ public class HbaseOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private String[] columnTypes;
     private Map<String, String> columnNameFamily;
 
+    private boolean kerberosAuthEnable;
+    private String regionserverKeytabFile;
+    private String regionserverPrincipal;
+    private String securityKrb5Conf;
+    private String zookeeperSaslClient;
+
     private String[] families;
     private String[] qualifiers;
 
@@ -252,6 +258,31 @@ public HbaseOutputFormatBuilder setColumnNameFamily(Map<String, String> columnNa
             format.columnNameFamily = columnNameFamily;
             return this;
         }
+        public HbaseOutputFormatBuilder setKerberosAuthEnable(boolean kerberosAuthEnable) {
+            format.kerberosAuthEnable = kerberosAuthEnable;
+            return this;
+        }
+
+        public HbaseOutputFormatBuilder setRegionserverKeytabFile(String regionserverKeytabFile) {
+            format.regionserverKeytabFile = regionserverKeytabFile;
+            return this;
+        }
+
+        public HbaseOutputFormatBuilder setRegionserverPrincipal(String regionserverPrincipal) {
+            format.regionserverPrincipal = regionserverPrincipal;
+            return this;
+        }
+
+        public HbaseOutputFormatBuilder setSecurityKrb5Conf(String securityKrb5Conf) {
+            format.securityKrb5Conf = securityKrb5Conf;
+            return this;
+        }
+
+        public HbaseOutputFormatBuilder setZookeeperSaslClient(String zookeeperSaslClient) {
+            format.zookeeperSaslClient = zookeeperSaslClient;
+            return this;
+        }
+
 
         public HbaseOutputFormat finish() {
             Preconditions.checkNotNull(format.host, "zookeeperQuorum should be specified");
@@ -267,7 +298,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];
                 }

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

@@ -52,6 +52,15 @@ public class HbaseSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
     protected String tableName;
     protected String updateMode;
     protected String rowkey;
+    protected String registerTabName;
+
+    protected boolean kerberosAuthEnable;
+    protected String regionserverKeytabFile;
+    protected String regionserverPrincipal;
+    protected String securityKrb5Conf;
+    protected String zookeeperSaslClient;
+    private int parallelism = -1;
+
 
     public HbaseSink() {
         // TO DO NOTHING
@@ -66,20 +75,34 @@ public HbaseSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.tableName = hbaseTableInfo.getTableName();
         this.rowkey = hbaseTableInfo.getRowkey();
         this.columnNameFamily = hbaseTableInfo.getColumnNameFamily();
-        this.updateMode = hbaseTableInfo.getUpdateMode();
+        this.registerTabName =  hbaseTableInfo.getName();
+
+        this.kerberosAuthEnable = hbaseTableInfo.isKerberosAuthEnable();
+        this.regionserverKeytabFile = hbaseTableInfo.getRegionserverKeytabFile();
+        this.regionserverPrincipal = hbaseTableInfo.getRegionserverPrincipal();
+        this.securityKrb5Conf = hbaseTableInfo.getSecurityKrb5Conf();
+        this.zookeeperSaslClient = hbaseTableInfo.getZookeeperSaslClient();
+
+        Integer tmpSinkParallelism = hbaseTableInfo.getParallelism();
+        if (tmpSinkParallelism != null) {
+            this.parallelism = tmpSinkParallelism;
+        }
         return this;
     }
 
     @Override
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         HbaseOutputFormat.HbaseOutputFormatBuilder builder = HbaseOutputFormat.buildHbaseOutputFormat();
-        builder.setHost(this.zookeeperQuorum)
-                .setZkParent(this.parent)
-                .setTable(this.tableName)
-                .setRowkey(rowkey)
-                .setUpdateMode(updateMode)
-                .setColumnNames(fieldNames)
-                .setColumnNameFamily(columnNameFamily);
+        builder.setHost(this.zookeeperQuorum).setZkParent(this.parent).setTable(this.tableName);
+
+        builder.setRowkey(rowkey);
+        builder.setColumnNames(fieldNames);
+        builder.setColumnNameFamily(columnNameFamily);
+        builder.setKerberosAuthEnable(kerberosAuthEnable);
+        builder.setRegionserverKeytabFile(regionserverKeytabFile);
+        builder.setRegionserverPrincipal(regionserverPrincipal);
+        builder.setSecurityKrb5Conf(securityKrb5Conf);
+        builder.setZookeeperSaslClient(zookeeperSaslClient);
 
         HbaseOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);