From 30a9424e19b8fa137b2fcb2a599a86eec4e5092d Mon Sep 17 00:00:00 2001
From: Woonduk Kang <wd.kang@navercorp.com>
Date: Fri, 10 Apr 2015 20:32:29 +0900
Subject: [PATCH] #320 change deprecated HTablePool api to HConnection

---
 .../common/hbase/PooledHTableFactory.java     | 52 +++++++++++++++----
 1 file changed, 41 insertions(+), 11 deletions(-)

diff --git a/commons/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java b/commons/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java
index ece9b5681..73b9f79c0 100644
--- a/commons/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java
+++ b/commons/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java
@@ -16,17 +16,22 @@
 
 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.
@@ -34,12 +39,24 @@ import java.util.concurrent.Executors;
  */
 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);
+        this(config, DEFAULT_POOL_SIZE, DEFAULT_WORKER_QUEUE_SIZE);
+    }
+    public PooledHTableFactory(Configuration config, int 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) {
@@ -47,13 +64,13 @@ public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBe
         }
     }
 
-    public PooledHTableFactory(Configuration config, int poolSize) {
-        this.executor = Executors.newFixedThreadPool(poolSize);
-        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;
     }
 
 
@@ -62,7 +79,7 @@ public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBe
         try {
             return connection.getTable(tableName, executor);
         } catch (IOException e) {
-            return null;
+            throw new HbaseSystemException(e);
         }
     }
 
@@ -79,5 +96,18 @@ public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBe
         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();
+            }
+        }
     }
 }