Merge pull request #161 from oceanbase/fix_put_comp_conn

Fix put comp conn

Author: JackShi148

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

@@ -122,13 +122,6 @@ public class OHTable implements HTableInterface {
      */
     private boolean               closeClientOnClose     = true;
 
-    /**
-     * If the connection this ObTable obtains is created by the ObTable itself,
-     * should set true and close the connection when this ObTable closes;
-     * otherwise set false
-     */
-    private final boolean         cleanupConnectionOnClose;
-
     /**
      * when the operationExecuteInPool is true the <code>Get</code>
      * will be executed in the pool.
@@ -166,11 +159,6 @@ public class OHTable implements HTableInterface {
 
     private int                   scannerTimeout;
 
-    /**
-     * the connection to obtain bufferedMutator for Put operations
-     */
-    private OHConnectionImpl      connection;
-
     /**
      * the bufferedMutator to execute Puts
      */
@@ -199,8 +187,6 @@ public OHTable(Configuration configuration, String tableName) throws IOException
         this.configuration = configuration;
         this.tableName = tableName.getBytes();
         this.tableNameString = tableName;
-        this.connection = (OHConnectionImpl) ConnectionFactory.createConnection(configuration);
-        this.cleanupConnectionOnClose = true;
 
         int maxThreads = configuration.getInt(HBASE_HTABLE_PRIVATE_THREADS_MAX,
             DEFAULT_HBASE_HTABLE_PRIVATE_THREADS_MAX);
@@ -258,8 +244,6 @@ public OHTable(Configuration configuration, final byte[] tableName,
         this.configuration = configuration;
         this.tableName = tableName;
         this.tableNameString = Bytes.toString(tableName);
-        this.connection = (OHConnectionImpl) ConnectionFactory.createConnection(configuration);
-        this.cleanupConnectionOnClose = true;
         this.executePool = executePool;
         this.cleanupPoolOnClose = false;
         OHConnectionConfiguration ohConnectionConf = new OHConnectionConfiguration(configuration);
@@ -296,7 +280,6 @@ public OHTable(final byte[] tableName, final ObTableClient obTableClient,
         this.tableNameString = Bytes.toString(tableName);
         this.cleanupPoolOnClose = false;
         this.closeClientOnClose = false;
-        this.cleanupConnectionOnClose = false;
         this.executePool = executePool;
         this.obTableClient = obTableClient;
         this.configuration = HBaseConfiguration.create();
@@ -314,8 +297,6 @@ public OHTable(TableName tableName, Connection connection,
         this.tableNameString = Bytes.toString(tableName.getName());
         this.configuration = connection.getConfiguration();
         this.executePool = executePool;
-        this.connection = (OHConnectionImpl) connection;
-        this.cleanupConnectionOnClose = false;
         if (executePool == null) {
             int maxThreads = configuration.getInt(HBASE_HTABLE_PRIVATE_THREADS_MAX,
                 DEFAULT_HBASE_HTABLE_PRIVATE_THREADS_MAX);
@@ -1531,11 +1512,6 @@ public void close() throws IOException {
         if (cleanupPoolOnClose) {
             executePool.shutdown();
         }
-        if (cleanupConnectionOnClose) {
-            if (this.connection != null) {
-                this.connection.close();
-            }
-        }
         this.isClosed = true;
     }
 
@@ -2262,10 +2238,9 @@ public Pair<byte[][], byte[][]> getStartEndKeys() throws IOException {
 
     private BufferedMutator getBufferedMutator() throws IOException {
         if (this.mutator == null) {
-            this.mutator = (OHBufferedMutatorImpl) this.connection.getBufferedMutator(
-                new BufferedMutatorParams(TableName.valueOf(this.tableNameString))
-                    .pool(this.executePool).writeBufferSize(this.writeBufferSize)
-                    .maxKeyValueSize(this.maxKeyValueSize), this);
+            this.mutator = new OHBufferedMutatorImpl(this.configuration, new BufferedMutatorParams(
+                TableName.valueOf(this.tableNameString)).pool(this.executePool)
+                .writeBufferSize(this.writeBufferSize).maxKeyValueSize(this.maxKeyValueSize), this);
         }
         return this.mutator;
     }

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

@@ -18,6 +18,7 @@
 package com.alipay.oceanbase.hbase.util;
 
 import com.alipay.oceanbase.hbase.OHTable;
+import com.alipay.oceanbase.rpc.exception.ObTableUnexpectedException;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableBatchOperation;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
@@ -87,6 +88,33 @@ public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParam
         }
     }
 
+    /**
+     * only used for OHTable get bufferedMutator
+     * */
+    public OHBufferedMutatorImpl(Configuration conf, BufferedMutatorParams params, OHTable ohTable)
+                                                                                                   throws IOException {
+        // create an OHTable object to do batch work
+        if (ohTable == null) {
+            throw new ObTableUnexpectedException("The ohTable is null.");
+        }
+        this.ohTable = ohTable;
+        // init params in OHBufferedMutatorImpl
+        this.tableName = params.getTableName();
+        this.conf = conf;
+        this.listener = params.getListener();
+
+        OHConnectionConfiguration connectionConfig = new OHConnectionConfiguration(conf);
+        this.pool = params.getPool();
+        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();
+
+    }
+
     @Override
     public TableName getName() {
         return this.tableName;