hbase support batchCallBack (oceanbase#86)

Author: stuBirdFly

src/main/java/com/alipay/oceanbase/hbase/OHTable.java

@@ -519,8 +519,18 @@ private String getTargetTableName(List<? extends Row> actions) {
     @Override
     public <R> void batchCallback(List<? extends Row> actions, Object[] results,
                                   Batch.Callback<R> callback) throws IOException,
-                                                             InterruptedException {
-        throw new FeatureNotSupportedException("not supported yet'");
+            InterruptedException {
+        try {
+            batch(actions, results);
+        } finally {
+            if (results != null) {
+                for (int i = 0; i < results.length; i++) {
+                    if (!(results[i] instanceof ObTableException)) {
+                        callback.update(null, actions.get(i).getRow(), (R) results[i]);
+                    }
+                }
+            }
+        }
     }
 
     public static int compareByteArray(byte[] bt1, byte[] bt2) {

src/test/java/com/alipay/oceanbase/hbase/OHTableMultiColumnFamilyTest.java

@@ -17,10 +17,11 @@
 
 package com.alipay.oceanbase.hbase;
 
+import com.alipay.oceanbase.rpc.mutation.result.MutationResult;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
 import org.apache.hadoop.hbase.filter.PrefixFilter;
 import org.junit.*;
 import org.junit.rules.ExpectedException;
@@ -152,6 +153,32 @@ public void testMultiColumnFamilyBatch() throws Exception {
         result = hTable.get(get);
         keyValues = result.rawCells();
         assertEquals(6, keyValues.length);
+
+        batchLsit.clear();
+        final long[] updateCounter = new long[] { 0L };
+        delete = new Delete(toBytes("Key5"));
+        delete.deleteColumns(family1, family1_column2);
+        delete.deleteColumns(family2, family2_column1);
+        delete.deleteFamily(family3);
+        batchLsit.add(delete);
+        for (int i = 0; i < rows; ++i) {
+            Put put = new Put(toBytes("Key" + i));
+            put.add(family1, family1_column1, family1_value);
+            put.add(family1, family1_column2, family1_value);
+            put.add(family1, family1_column3, family1_value);
+            put.add(family2, family2_column1, family2_value);
+            put.add(family2, family2_column2, family2_value);
+            put.add(family3, family3_column1, family3_value);
+            batchLsit.add(put);
+        }
+        hTable.batchCallback(batchLsit, new Batch.Callback<MutationResult>() {
+            @Override
+            public void update(byte[] region, byte[] row, MutationResult result) {
+                updateCounter[0]++;
+            }
+        });
+        assertEquals(11, updateCounter[0]);
+
     }
 
     @Test