resolve conflict

Author: simenliuxing

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

@@ -20,6 +20,7 @@
 
 package com.dtstack.flink.sql.sink.hbase;
 
+import com.dtstack.flink.sql.enums.EUpdateMode;
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
@@ -30,8 +31,10 @@
 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;
 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;
@@ -150,8 +153,13 @@ public Connection run() {
     @Override
     public void writeRecord(Tuple2 tuple2) {
         Tuple2<Boolean, Row> tupleTrans = tuple2;
+        Boolean retract = tupleTrans.f0;
         Row row = tupleTrans.f1;
-        dealInsert(row);
+        if (retract) {
+            dealInsert(row);
+        } else if (!retract && StringUtils.equalsIgnoreCase(updateMode, EUpdateMode.UPSERT.name())) {
+            dealDelete(row);
+        }
     }
 
     protected void dealInsert(Row record) {
@@ -177,6 +185,26 @@ protected void dealInsert(Row record) {
         outRecords.inc();
     }
 
+    protected void dealDelete(Row record) {
+        String rowKey = buildRowKey(record);
+        if (!StringUtils.isEmpty(rowKey)) {
+            Delete delete = new Delete(Bytes.toBytes(rowKey));
+            try {
+                table.delete(delete);
+            } catch (IOException e) {
+                if (outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0 || LOG.isDebugEnabled()) {
+                    LOG.error("record insert failed ..{}", record.toString());
+                    LOG.error("", e);
+                }
+                outDirtyRecords.inc();
+            }
+            if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
+                LOG.info(record.toString());
+            }
+            outRecords.inc();
+        }
+    }
+
     private Put getPutByRow(Row record) {
         String rowKey = buildRowKey(record);
         if (StringUtils.isEmpty(rowKey)) {
@@ -187,7 +215,7 @@ private Put getPutByRow(Row record) {
             Object fieldVal = record.getField(i);
             byte[] val = null;
             if (fieldVal != null) {
-                val = HbaseUtil.toByte(fieldVal);
+                val = fieldVal.toString().getBytes();
             }
             byte[] cf = families[i].getBytes();
             byte[] qualifier = qualifiers[i].getBytes();