BufferedMutator 1.3 batch (#88)

* add DepentdentFilter and SingleColumnValueExcludeFilter

* add singleColumnValueExcludeFilter and DependentColumnFilter; add test cases and optimize bufferedMutator test cases

* remove DeleteFamilyVersion file and move all cases to MultiColumnFamilyTest

* adjust bufferdMutatorImpl 1.x to new batch

* bufferedMutator do not retry, batch retry in table client

* add rpcTimeout and operationTimetout setting in bufferedMutator

* clean execBuffer if all commits have succeed

* format code

* change connectionConfig to local variable; add deprecated interfaces in 1.x

* add check null

Author: JackShi148

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

@@ -198,8 +198,13 @@ public OHTable(Configuration configuration, String tableName) throws IOException
             DEFAULT_HBASE_HTABLE_THREAD_KEEP_ALIVE_TIME);
         this.executePool = createDefaultThreadPoolExecutor(1, maxThreads, keepAliveTime);
         OHConnectionConfiguration ohConnectionConf = new OHConnectionConfiguration(configuration);
+        int numRetries = configuration.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+                HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
         this.obTableClient = ObTableClientManager.getOrCreateObTableClient(setUserDefinedNamespace(
             this.tableNameString, ohConnectionConf));
+        this.obTableClient.setRpcExecuteTimeout(ohConnectionConf.getRpcTimeout());
+        this.obTableClient.setRuntimeRetryTimes(numRetries);
+        setOperationTimeout(ohConnectionConf.getOperationTimeout());
 
         finishSetUp();
     }
@@ -246,8 +251,13 @@ public OHTable(Configuration configuration, final byte[] tableName,
         this.executePool = executePool;
         this.cleanupPoolOnClose = false;
         OHConnectionConfiguration ohConnectionConf = new OHConnectionConfiguration(configuration);
+        int numRetries = configuration.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+                HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
         this.obTableClient = ObTableClientManager.getOrCreateObTableClient(setUserDefinedNamespace(
             this.tableNameString, ohConnectionConf));
+        this.obTableClient.setRpcExecuteTimeout(ohConnectionConf.getRpcTimeout());
+        this.obTableClient.setRuntimeRetryTimes(numRetries);
+        setOperationTimeout(ohConnectionConf.getOperationTimeout());
 
         finishSetUp();
     }
@@ -301,6 +311,7 @@ public OHTable(TableName tableName, Connection connection,
         } else {
             this.cleanupPoolOnClose = false;
         }
+
         this.rpcTimeout = connectionConfig.getRpcTimeout();
         this.operationTimeout = connectionConfig.getOperationTimeout();
         this.operationExecuteInPool = this.configuration.getBoolean(
@@ -311,8 +322,15 @@ public OHTable(TableName tableName, Connection connection,
             DEFAULT_HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK);
         this.writeBufferSize = connectionConfig.getWriteBufferSize();
         this.tableName = tableName.getName();
+        int numRetries = configuration.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+                HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
         this.obTableClient = ObTableClientManager.getOrCreateObTableClient(setUserDefinedNamespace(
             this.tableNameString, connectionConfig));
+        this.obTableClient.setRpcExecuteTimeout(rpcTimeout);
+        this.obTableClient.setRuntimeRetryTimes(numRetries);
+        setOperationTimeout(operationTimeout);
+
+        finishSetUp();
     }
 
     /**

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

@@ -18,14 +18,11 @@
 package com.alipay.oceanbase.hbase.util;
 
 import com.alipay.oceanbase.hbase.OHTable;
-import com.alipay.oceanbase.rpc.ObTableClient;
-import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.*;
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.*;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.slf4j.Logger;
 
 import java.io.IOException;
@@ -34,54 +31,53 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 import static com.alipay.oceanbase.rpc.util.TableClientLoggerFactory.LCD;
 
 @InterfaceAudience.Private
 public class OHBufferedMutatorImpl implements BufferedMutator {
-    private static final Logger             LOGGER                 = TableHBaseLoggerFactory
-                                                                       .getLogger(OHBufferedMutatorImpl.class);
+    private static final Logger           LOGGER                 = TableHBaseLoggerFactory
+                                                                     .getLogger(OHBufferedMutatorImpl.class);
 
-    private final ExceptionListener         listener;
+    private final ExceptionListener       listener;
 
-    protected final ObTableClient           obTableClient;
-    private final TableName                 tableName;
-    private volatile Configuration          conf;
-    private final OHConnectionConfiguration connectionConfig;
+    private final OHTable                 ohTable;
+    private final TableName               tableName;
+    private volatile Configuration        conf;
 
-    final ConcurrentLinkedQueue<Mutation>   asyncWriteBuffer       = new ConcurrentLinkedQueue<Mutation>();
-    AtomicLong                              currentAsyncBufferSize = new AtomicLong(0);
+    @VisibleForTesting
+    final ConcurrentLinkedQueue<Mutation> asyncWriteBuffer       = new ConcurrentLinkedQueue<Mutation>();
+    @VisibleForTesting
+    AtomicLong                            currentAsyncBufferSize = new AtomicLong(0);
 
-    private AtomicReference<Class<?>>       type                   = new AtomicReference<>(null);
-    private final long                      writeBufferSize;
-    private final int                       maxKeyValueSize;
-    private boolean                         closed                 = false;
-    private final ExecutorService           pool;
-    private final int                       rpcTimeout;
+    private long                          writeBufferSize;
+    private final int                     maxKeyValueSize;
+    private boolean                       closed                 = false;
+    private final ExecutorService         pool;
+    private int                           rpcTimeout;
+    private int                           operationTimeout;
 
     public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParams params)
                                                                                              throws IOException {
         if (ohConnection == null || ohConnection.isClosed()) {
             throw new IllegalArgumentException("Connection is null or closed.");
         }
-        // create a ObTableClient to do rpc operations
-        this.obTableClient = ObTableClientManager.getOrCreateObTableClient(ohConnection
-            .getOHConnectionConfiguration());
-
         // init params in OHBufferedMutatorImpl
         this.tableName = params.getTableName();
         this.conf = ohConnection.getConfiguration();
-        this.connectionConfig = ohConnection.getOHConnectionConfiguration();
         this.listener = params.getListener();
+
+        OHConnectionConfiguration connectionConfig = ohConnection.getOHConnectionConfiguration();
         this.pool = params.getPool();
-        this.obTableClient.setRuntimeBatchExecutor(pool);
+        this.rpcTimeout = connectionConfig.getRpcTimeout();
+        this.operationTimeout = connectionConfig.getOperationTimeout();
 
         this.writeBufferSize = params.getWriteBufferSize() != OHConnectionImpl.BUFFERED_PARAM_UNSET ? params
             .getWriteBufferSize() : connectionConfig.getWriteBufferSize();
         this.maxKeyValueSize = params.getMaxKeyValueSize() != OHConnectionImpl.BUFFERED_PARAM_UNSET ? params
             .getMaxKeyValueSize() : connectionConfig.getMaxKeyValueSize();
-        this.rpcTimeout = connectionConfig.getRpcTimeout();
-        this.obTableClient.setRpcExecuteTimeout(rpcTimeout);
+
+        // create an OHTable object to do batch work
+        this.ohTable = new OHTable(tableName, ohConnection, connectionConfig, pool);
     }
 
     @Override
@@ -119,38 +115,37 @@ public void mutate(List<? extends Mutation> mutations) throws IOException {
         }
 
         long toAddSize = 0;
-        // check if every mutation's family is the same
-        // check if mutations are the same type
         for (Mutation m : mutations) {
-            OHTable.checkFamilyViolation(m.getFamilyMap().keySet(), true);
-            validateInsUpAndDelete(m);
-            Class<?> curType = m.getClass();
-            // set the type of this BufferedMutator
-            if (type.get() == null) {
-                type.compareAndSet(null, mutations.get(0).getClass());
-            }
-            if (!type.get().equals(curType)) {
-                throw new IllegalArgumentException("Not support different type in one batch.");
-            }
+            validateOperation(m);
             toAddSize += m.heapSize();
         }
 
         currentAsyncBufferSize.addAndGet(toAddSize);
         asyncWriteBuffer.addAll(mutations);
 
-        asyncExecute(false);
+        if (currentAsyncBufferSize.get() > writeBufferSize) {
+            execute(false);
+        }
+
     }
 
     /**
      * Check whether the mutation is Put or Delete in 1.x
      * @param mt - mutation operation
      */
-    private void validateInsUpAndDelete(Mutation mt) throws IllegalArgumentException {
+    private void validateOperation(Mutation mt) throws IllegalArgumentException {
+        if (mt == null) {
+            throw new IllegalArgumentException("Mutation operation cannot be null");
+        }
         if (!(mt instanceof Put) && !(mt instanceof Delete)) {
             throw new IllegalArgumentException("Only support for Put and Delete for now.");
         }
         if (mt instanceof Put) {
+            // family empty check is in validatePut
             HTable.validatePut((Put) mt, maxKeyValueSize);
+            OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), true);
+        } else {
+            OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), false);
         }
     }
 
@@ -161,91 +156,49 @@ private void validateInsUpAndDelete(Mutation mt) throws IllegalArgumentException
      * @param flushAll - if true, sends all the writes and wait for all of them to finish before
      *        returning.
      */
-    private void asyncExecute(boolean flushAll) throws IOException {
+    private void execute(boolean flushAll) throws IOException {
         LinkedList<Mutation> execBuffer = new LinkedList<>();
-        ObTableBatchOperationRequest request = null;
-        // namespace n1, n1:table_name
-        // namespace default, table_name
-        String tableNameString = tableName.getNameAsString();
+        long dequeuedSize = 0L;
         try {
-            while (true) {
-                try{
-                    if (!flushAll || asyncWriteBuffer.isEmpty()) {
-                        if (currentAsyncBufferSize.get() <= writeBufferSize) {
-                            break;
-                        }
-                    }
-                    Mutation m;
-                    while ((m = asyncWriteBuffer.poll()) != null) {
-                        execBuffer.add(m);
-                        long size = m.heapSize();
-                        currentAsyncBufferSize.addAndGet(-size);
-                    }
-                    // in concurrent situation, asyncWriteBuffer may be empty here
-                    // for other threads flush all buffer
-                    if (execBuffer.isEmpty()) {
-                        break;
-                    }
-                    // for now, operations' family is the same
-                    byte[] family = execBuffer.getFirst().getFamilyMap().firstKey();
-                    ObTableBatchOperation batch = buildObTableBatchOperation(execBuffer);
-                    // table_name$cf_name
-                    String targetTableName = OHTable.getTargetTableName(tableNameString, Bytes.toString(family), conf);
-                    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);
-                }
-                try {
-                    ObTableBatchOperationResult result = (ObTableBatchOperationResult) obTableClient.execute(request);
-                } catch (Exception ex) {
-                    LOGGER.debug("Errors occur during mutation operation", ex);
-                    Mutation m = null;
-                    try {
-                        // retry every single operation
-                        while (!execBuffer.isEmpty()) {
-                            // poll elements from execBuffer to recollect remaining operations
-                            m = execBuffer.poll();
-                            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);
-                            ObTableBatchOperationResult result = (ObTableBatchOperationResult) obTableClient.execute(request);
-                        }
-                    } catch (Exception newEx) {
-                        if (m != null) {
-                            execBuffer.addFirst(m);
-                        }
-                        // if retry fails, only recollect remaining operations
-                        while(!execBuffer.isEmpty()) {
-                            m = execBuffer.poll();
-                            long size = m.heapSize();
-                            asyncWriteBuffer.add(m);
-                            currentAsyncBufferSize.addAndGet(size);
-                        }
-                        throw newEx;
-                    }
-                }
+            Mutation m;
+            while ((writeBufferSize <= 0 || dequeuedSize < (writeBufferSize * 2) || flushAll)
+                && (m = asyncWriteBuffer.poll()) != null) {
+                execBuffer.add(m);
+                long size = m.heapSize();
+                currentAsyncBufferSize.addAndGet(-size);
+                dequeuedSize += size;
+            }
+
+            if (execBuffer.isEmpty()) {
+                return;
             }
+            ohTable.batch(execBuffer);
+            // if commit all successfully, clean execBuffer
+            execBuffer.clear();
         } catch (Exception ex) {
             LOGGER.error(LCD.convert("01-00026"), ex);
-            // if the cause is illegal argument, directly throw to user
-            if (ex instanceof IllegalArgumentException) {
-                throw (IllegalArgumentException) ex;
-            }
-            // TODO: need to collect error information and actions during batch operations
-            // TODO: maybe keep in ObTableBatchOperationResult
-            List<Throwable> throwables = new ArrayList<Throwable>();
-            List<Row> actions = new ArrayList<Row>();
-            List<String> addresses = new ArrayList<String>();
-            throwables.add(ex);
-            RetriesExhaustedWithDetailsException error = new RetriesExhaustedWithDetailsException(
-                    new ArrayList<Throwable>(throwables),
-                    new ArrayList<Row>(actions), new ArrayList<String>(addresses));
-            if (listener == null) {
-                throw error;
+            if (ex.getCause() instanceof RetriesExhaustedWithDetailsException) {
+                LOGGER.error(tableName + ": One or more of the operations have failed after retries.");
+                RetriesExhaustedWithDetailsException retryException = (RetriesExhaustedWithDetailsException) ex.getCause();
+                // recollect mutations
+                execBuffer.clear();
+                for (int i = 0; i < retryException.getNumExceptions(); ++i) {
+                    execBuffer.add((Mutation) retryException.getRow(i));
+                }
+                if (listener != null) {
+                    listener.onException(retryException, this);
+                } else {
+                    throw retryException;
+                }
             } else {
-                listener.onException(error, this);
+                LOGGER.error("Errors unrelated to operations occur during mutation operation", ex);
+                throw ex;
+            }
+        } finally {
+            for (Mutation mutation : execBuffer) {
+                long size = mutation.heapSize();
+                currentAsyncBufferSize.addAndGet(size);
+                asyncWriteBuffer.add(mutation);
             }
         }
     }
@@ -256,7 +209,7 @@ public void close() throws IOException {
             return;
         }
         try {
-            asyncExecute(true);
+            execute(true);
         } finally {
             // the pool in ObTableClient will be shut down too
             this.pool.shutdown();
@@ -273,27 +226,40 @@ public void close() throws IOException {
         }
     }
 
+    @Deprecated
+    public void setWriteBufferSize(long writeBufferSize) throws IOException {
+        this.writeBufferSize = writeBufferSize;
+        if (currentAsyncBufferSize.get() > writeBufferSize) {
+            flush();
+        }
+    }
+
     /**
      * Force to commit all operations
      * do not care whether the pool is shut down or this BufferedMutator is closed
      */
     @Override
     public void flush() throws IOException {
-        asyncExecute(true);
+        execute(true);
     }
 
     @Override
     public long getWriteBufferSize() {
         return this.writeBufferSize;
     }
 
-    private ObTableBatchOperation buildObTableBatchOperation(List<? extends Mutation> execBuffer) {
-        List<KeyValue> keyValueList = new LinkedList<>();
-        for (Mutation mutation : execBuffer) {
-            for (Map.Entry<byte[], List<KeyValue>> entry : mutation.getFamilyMap().entrySet()) {
-                keyValueList.addAll(entry.getValue());
-            }
-        }
-        return OHTable.buildObTableBatchOperation(keyValueList, false, null);
+    public void setRpcTimeout(int rpcTimeout) {
+        this.rpcTimeout = rpcTimeout;
+        this.ohTable.setRpcTimeout(rpcTimeout);
+    }
+
+    public void setOperationTimeout(int operationTimeout) {
+        this.operationTimeout = operationTimeout;
+        this.ohTable.setOperationTimeout(operationTimeout);
+    }
+
+    @Deprecated
+    public List<Row> getWriteBuffer() {
+        return Arrays.asList(asyncWriteBuffer.toArray(new Row[0]));
     }
 }