From 114764548d85b3742fd4a8bd3b20b8752d251b1a Mon Sep 17 00:00:00 2001
From: stuBirdFly <1065492934@qq.com>
Date: Wed, 23 Oct 2024 11:31:44 +0800
Subject: [PATCH] hbase support batchCallBack

---
 .../com/alipay/oceanbase/hbase/OHTable.java   | 23 +++++++++++----
 .../hbase/OHTableMultiColumnFamilyTest.java   | 28 +++++++++++++++++++
 2 files changed, 45 insertions(+), 6 deletions(-)

diff --git a/src/main/java/com/alipay/oceanbase/hbase/OHTable.java b/src/main/java/com/alipay/oceanbase/hbase/OHTable.java
index 6d1cc873..4712527b 100644
--- a/src/main/java/com/alipay/oceanbase/hbase/OHTable.java
+++ b/src/main/java/com/alipay/oceanbase/hbase/OHTable.java
@@ -526,15 +526,26 @@ public Object[] batch(List<? extends Row> actions) throws IOException {
     @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]);
+                    }
+                }
+            }
+        }
     }
 
     @Override
-    public <R> Object[] batchCallback(List<? extends Row> actions, Batch.Callback<R> callback)
-                                                                                              throws IOException,
-                                                                                              InterruptedException {
-        throw new FeatureNotSupportedException("not supported yet'");
+    public <R> Object[] batchCallback(
+            final List<? extends Row> actions, final Batch.Callback<R> callback) throws IOException, InterruptedException {
+        Object[] results = new Object[actions.size()];
+        batchCallback(actions, results, callback);
+        return results;
     }
 
     public static int compareByteArray(byte[] bt1, byte[] bt2) {
diff --git a/src/test/java/com/alipay/oceanbase/hbase/OHTableMultiColumnFamilyTest.java b/src/test/java/com/alipay/oceanbase/hbase/OHTableMultiColumnFamilyTest.java
index f8d4eed3..a649e58e 100644
--- a/src/test/java/com/alipay/oceanbase/hbase/OHTableMultiColumnFamilyTest.java
+++ b/src/test/java/com/alipay/oceanbase/hbase/OHTableMultiColumnFamilyTest.java
@@ -17,8 +17,10 @@
 
 package com.alipay.oceanbase.hbase;
 
+import com.alipay.oceanbase.rpc.mutation.result.MutationResult;
 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;
@@ -146,6 +148,32 @@ public void testMultiColumnFamilyBatch() throws Exception {
         result = hTable.get(get);
         keyValues = result.raw();
         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