OHBufferedMutator set and add runtimeBatchExecutor in ObTableClient (#68)

* init bufferedMutator

* finish validateFamily and asyncExecute

* correct log in OHBufferedMutatorImpl

* pass self-test

* format code

* add retry when batch fails

* remove test print

* format code

* make interface more generalized

* format BufferedMutator test case

* remove redundancy, add some comments

* fix type of a bufferedMutator. Optimize by review

* OHBufferedMutator set and use runtimeBatchExecutor in ObTableClient

Author: JackShi148

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

@@ -545,7 +545,8 @@ public Result call() throws IOException {
                             family = entry.getKey();
                             obTableQuery = buildObTableQuery(get, entry.getValue());
                             request = buildObTableQueryAsyncRequest(obTableQuery,
-                                getTargetTableName(tableNameString, Bytes.toString(family)));
+                                getTargetTableName(tableNameString, Bytes.toString(family),
+                                    configuration));
                             clientQueryStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
                                 .execute(request);
                             getKeyValueFromResult(clientQueryStreamResult, keyValueList, false,
@@ -810,7 +811,7 @@ private void innerDelete(Delete delete) throws IOException {
                         .next();
 
                 BatchOperation batch = buildBatchOperation(
-                        getTargetTableName(tableNameString, Bytes.toString(entry.getKey())),
+                        getTargetTableName(tableNameString, Bytes.toString(entry.getKey()), configuration),
                         entry.getValue(), false, null);
                 results = batch.execute();
             }
@@ -1729,14 +1730,12 @@ private ObTableQueryAsyncRequest buildObTableQueryAsyncRequest(ObTableQuery obTa
     }
 
     public static ObTableBatchOperationRequest buildObTableBatchOperationRequest(ObTableBatchOperation obTableBatchOperation,
-                                                                                 String targetTableName,
-                                                                                 ExecutorService pool) {
+                                                                                 String targetTableName) {
         ObTableBatchOperationRequest request = new ObTableBatchOperationRequest();
         request.setTableName(targetTableName);
         request.setReturningAffectedRows(true);
         request.setEntityType(ObTableEntityType.HKV);
         request.setBatchOperation(obTableBatchOperation);
-        request.setPool(pool);
         return request;
     }
 
@@ -1754,7 +1753,7 @@ private ObTableQueryAndMutateRequest buildObTableQueryAndMutateRequest(ObTableQu
         return request;
     }
 
-    private void checkFamilyViolation(Collection<byte[]> families) {
+    public static void checkFamilyViolation(Collection<byte[]> families) {
         for (byte[] family : families) {
             if (isBlank(Bytes.toString(family))) {
                 throw new IllegalArgumentException("family is blank");

src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java

@@ -88,13 +88,13 @@ public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParam
         this.obTableClient = ObTableClientManager.getOrCreateObTableClient(ohConnection
             .getOHConnectionConfiguration());
 
-        // init params in OHBufferedMutatorImpl:
-        // TableName + pool + Configuration + listener + writeBufferSize + maxKeyValueSize + rpcTimeout + operationTimeout
+        // init params in OHBufferedMutatorImpl
         this.tableName = params.getTableName();
         this.conf = ohConnection.getConfiguration();
         this.connectionConfig = ohConnection.getOHConnectionConfiguration();
         this.listener = params.getListener();
         this.pool = params.getPool();
+        this.obTableClient.setRuntimeBatchExecutor(pool);
 
         this.writeBufferSize = params.getWriteBufferSize() != OHConnectionImpl.BUFFERED_PARAM_UNSET ? params
             .getWriteBufferSize() : connectionConfig.getWriteBufferSize();
@@ -211,7 +211,7 @@ private void asyncExecute(boolean flushAll) throws IOException {
                     ObTableBatchOperation batch = buildObTableBatchOperation(execBuffer);
                     // table_name$cf_name
                     String targetTableName = OHTable.getTargetTableName(tableNameString, Bytes.toString(family), conf);
-                    request = OHTable.buildObTableBatchOperationRequest(batch, targetTableName, pool);
+                    request = OHTable.buildObTableBatchOperationRequest(batch, targetTableName);
                 } catch (Exception ex) {
                     LOGGER.error("Errors occur before mutation operation", ex);
                     throw new IllegalArgumentException("Errors occur before mutation operation", ex);
@@ -229,7 +229,7 @@ private void asyncExecute(boolean flushAll) throws IOException {
                             byte[] family = m.getFamilyMap().firstKey();
                             ObTableBatchOperation batch = buildObTableBatchOperation(Collections.singletonList(m));
                             String targetTableName = OHTable.getTargetTableName(tableNameString, Bytes.toString(family), conf);
-                            request = OHTable.buildObTableBatchOperationRequest(batch, targetTableName, pool);
+                            request = OHTable.buildObTableBatchOperationRequest(batch, targetTableName);
                             ObTableBatchOperationResult result = (ObTableBatchOperationResult) obTableClient.execute(request);
                         }
                     } catch (Exception newEx) {
@@ -278,6 +278,7 @@ public void close() throws IOException {
         try {
             asyncExecute(true);
         } finally {
+            // the pool in ObTableClient will be shut down too
             this.pool.shutdown();
             try {
                 if (!pool.awaitTermination(600, TimeUnit.SECONDS)) {