Merge pull request #321 from emeroad/#320_change_deprecated_HTablePoo…

…l_api_to_HConnection

#320 change deprecated HTablePool api to HConnection

commons/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java

@@ -16,53 +16,70 @@
 
 package com.navercorp.pinpoint.common.hbase;
 
+import com.navercorp.pinpoint.common.util.ExecutorFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.client.HTableInterface;
 import org.apache.hadoop.hbase.client.HTableInterfaceFactory;
 import org.apache.hadoop.hbase.client.HConnection;
 import org.apache.hadoop.hbase.client.HConnectionManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.data.hadoop.hbase.HbaseSystemException;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * HTableInterfaceFactory based on HTablePool.
  * @author emeroad
  */
 public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBean {
 
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    public static final int DEFAULT_POOL_SIZE = 256;
+    public static final int DEFAULT_WORKER_QUEUE_SIZE = 1024*5;
+
     private ExecutorService executor;
     private HConnection connection;
-    public static final int DEFAULT_POOL_SIZE = 256;
+
 
     public PooledHTableFactory(Configuration config) {
-        this.executor = Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
-        try {
-            this.connection = HConnectionManager.createConnection(config, executor);
-        } catch (IOException e) {
-            throw new HbaseSystemException(e);
-        }
+        this(config, DEFAULT_POOL_SIZE, DEFAULT_WORKER_QUEUE_SIZE);
     }
-
     public PooledHTableFactory(Configuration config, int poolSize) {
-        this.executor = Executors.newFixedThreadPool(poolSize);
+        this(config, poolSize, DEFAULT_WORKER_QUEUE_SIZE);
+    }
+
+    public PooledHTableFactory(Configuration config, int poolSize, int workerQueueSize) {
+        this.executor = getExecutorService(poolSize, workerQueueSize);
         try {
             this.connection = HConnectionManager.createConnection(config, executor);
         } catch (IOException e) {
             throw new HbaseSystemException(e);
         }
     }
 
+    private ExecutorService getExecutorService(int poolSize, int workQueueMaxSize) {
+
+        logger.info("create HConnectionThreadPoolExecutor poolSize:{}, workerQueueMaxSize:{}", poolSize, workQueueMaxSize);
+
+        ThreadPoolExecutor threadPoolExecutor = ExecutorFactory.newFixedThreadPool(poolSize, workQueueMaxSize, "Pinpoint-HConnectionExecutor", true);
+        threadPoolExecutor.prestartAllCoreThreads();
+        return threadPoolExecutor;
+    }
+
 
     @Override
     public HTableInterface createHTableInterface(Configuration config, byte[] tableName) {
         try {
             return connection.getTable(tableName, executor);
         } catch (IOException e) {
-            return null;
+            throw new HbaseSystemException(e);
         }
     }
 
@@ -79,5 +96,18 @@ public void destroy() throws Exception {
         if (connection != null) {
             this.connection.close();
         }
+
+        if (this.executor != null) {
+            this.executor.shutdown();
+            try {
+                final boolean shutdown = executor.awaitTermination(1000 * 5, TimeUnit.MILLISECONDS);
+                if (!shutdown) {
+                    final List<Runnable> discardTask = this.executor.shutdownNow();
+                    logger.warn("discard task size:{}", discardTask.size());
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
     }
 }