diff --git a/collector/src/main/resources/applicationContext-hbase.xml b/collector/src/main/resources/applicationContext-hbase.xml
index 968adea4c..8adfa0dae 100644
--- a/collector/src/main/resources/applicationContext-hbase.xml
+++ b/collector/src/main/resources/applicationContext-hbase.xml
@@ -19,18 +19,33 @@
${hbase.client.host}
${hbase.client.port}
- ${hbase.htable.threads.max}
- true
- 30000
- 10000
+
+
+ ${hbase.ipc.client.tcpnodelay}
+
+ ${hbase.rpc.timeout}
+
+ ${hbase.client.operation.timeout}
+
+
+ ${hbase.ipc.client.socket.timeout.read}
+
+ ${hbase.ipc.client.socket.timeout.write}
+
+
+
+
+
+
+
-
+
diff --git a/collector/src/main/resources/hbase.properties b/collector/src/main/resources/hbase.properties
index 3889f74b0..6f8afd78c 100644
--- a/collector/src/main/resources/hbase.properties
+++ b/collector/src/main/resources/hbase.properties
@@ -1,3 +1,22 @@
hbase.client.host=localhost
hbase.client.port=2181
-hbase.htable.threads.max=4
\ No newline at end of file
+
+# hbase timeout option==================================================================================
+# hbase default:true
+hbase.ipc.client.tcpnodelay=true
+# hbase default:60000
+hbase.rpc.timeout=10000
+# hbase default:Integer.MAX_VALUE
+hbase.client.operation.timeout=10000
+
+# hbase socket read timeout. default: 200000
+hbase.ipc.client.socket.timeout.read=20000
+# socket write timeout. hbase default: 600000
+hbase.ipc.client.socket.timeout.write=60000
+
+# ==================================================================================
+# hbase client thread pool option
+hbase.client.thread.max=1024
+hbase.client.threadPool.queueSize=5120
+# prestartAllCoreThreads
+hbase.client.threadPool.prestart=false
\ No newline at end of file
diff --git a/collector/src/main/resources/pinpoint-collector.properties b/collector/src/main/resources/pinpoint-collector.properties
index c61e7c484..53e6bb779 100644
--- a/collector/src/main/resources/pinpoint-collector.properties
+++ b/collector/src/main/resources/pinpoint-collector.properties
@@ -1,5 +1,3 @@
-hbase.hTablePoolSize=1024
-
# tcp listen ip
collector.tcpListenIp=0.0.0.0
collector.tcpListenPort=9994
diff --git a/commons/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java b/commons/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java
index c4e14aafe..eafe7f434 100644
--- a/commons/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java
+++ b/commons/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java
@@ -41,10 +41,7 @@ public class HbaseTemplate2 extends HbaseTemplate implements HbaseOperations2, I
private final Logger logger = LoggerFactory.getLogger(this.getClass());
- private PooledHTableFactory pooledHTableFactory;
- private int poolSize = PooledHTableFactory.DEFAULT_POOL_SIZE;
-
- private ExecutorService executor = newCachedThreadPool();
+ private final ExecutorService executor = newCachedThreadPool();
public HbaseTemplate2() {
}
@@ -65,35 +62,21 @@ public class HbaseTemplate2 extends HbaseTemplate implements HbaseOperations2, I
public HbaseTemplate2(Configuration configuration) {
Assert.notNull(configuration);
+ setConfiguration(configuration);
}
- public HbaseTemplate2(Configuration configuration, int poolSize) {
- Assert.notNull(configuration);
- this.poolSize = poolSize;
- }
- public int getPoolSize() {
- return poolSize;
- }
-
- public void setPoolSize(int hTablePoolSize) {
- this.poolSize = hTablePoolSize;
- }
@Override
public void afterPropertiesSet() {
Configuration configuration = getConfiguration();
Assert.notNull(configuration, "configuration is required");
- this.pooledHTableFactory = new PooledHTableFactory(configuration, poolSize);
- this.setTableFactory(pooledHTableFactory);
+ Assert.notNull(getTableFactory(), "tableFactory is required");
}
@Override
public void destroy() throws Exception {
- if (pooledHTableFactory != null) {
- this.pooledHTableFactory.destroy();
- }
-
+ logger.info("HbaseTemplate2.destroy()");
final ExecutorService executor = this.executor;
if (executor != null) {
executor.shutdown();
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 dc63770b1..a447a9802 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,13 +16,19 @@
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.HTablePool;
+import org.apache.hadoop.hbase.client.*;
+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.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
/**
* HTableInterfaceFactory based on HTablePool.
@@ -30,21 +36,50 @@ import java.io.IOException;
*/
public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBean {
- private HTablePool hTablePool;
+ 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;
+ public static final boolean DEFAULT_PRESTART_THREAD_POOL = false;
+
+ private final ExecutorService executor;
+ private final HConnection connection;
+
public PooledHTableFactory(Configuration config) {
- this.hTablePool = new HTablePool(config, DEFAULT_POOL_SIZE);
+ this(config, DEFAULT_POOL_SIZE, DEFAULT_WORKER_QUEUE_SIZE, DEFAULT_PRESTART_THREAD_POOL);
}
- public PooledHTableFactory(Configuration config, int poolSize) {
- this.hTablePool = new HTablePool(config, poolSize);
+ public PooledHTableFactory(Configuration config, int poolSize, int workerQueueSize, boolean prestartThreadPool) {
+ this.executor = createExecutorService(poolSize, workerQueueSize, prestartThreadPool);
+ try {
+ this.connection = (HConnection)ConnectionFactory.createConnection(config, executor);
+ } catch (IOException e) {
+ throw new HbaseSystemException(e);
+ }
+ }
+
+ private ExecutorService createExecutorService(int poolSize, int workQueueMaxSize, boolean prestartThreadPool) {
+
+ logger.info("create HConnectionThreadPoolExecutor poolSize:{}, workerQueueMaxSize:{}", poolSize, workQueueMaxSize);
+
+ ThreadPoolExecutor threadPoolExecutor = ExecutorFactory.newFixedThreadPool(poolSize, workQueueMaxSize, "Pinpoint-HConnectionExecutor", true);
+ if (prestartThreadPool) {
+ logger.info("prestartAllCoreThreads");
+ threadPoolExecutor.prestartAllCoreThreads();
+ }
+
+ return threadPoolExecutor;
}
@Override
public HTableInterface createHTableInterface(Configuration config, byte[] tableName) {
- return hTablePool.getTable(tableName);
+ try {
+ return connection.getTable(tableName, executor);
+ } catch (IOException e) {
+ throw new HbaseSystemException(e);
+ }
}
@Override
@@ -57,8 +92,26 @@ public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBe
@Override
public void destroy() throws Exception {
- if (hTablePool != null) {
- this.hTablePool.close();
+ logger.info("PooledHTableFactory.destroy()");
+ if (connection != null) {
+ try {
+ this.connection.close();
+ } catch (IOException ex) {
+ logger.warn("Connection.close() error:" + ex.getMessage(), ex);
+ }
+ }
+
+ if (this.executor != null) {
+ this.executor.shutdown();
+ try {
+ final boolean shutdown = executor.awaitTermination(1000 * 5, TimeUnit.MILLISECONDS);
+ if (!shutdown) {
+ final List discardTask = this.executor.shutdownNow();
+ logger.warn("discard task size:{}", discardTask.size());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
}
}
}
diff --git a/web/src/main/resources/applicationContext-hbase.xml b/web/src/main/resources/applicationContext-hbase.xml
index 82381f440..709d2e675 100644
--- a/web/src/main/resources/applicationContext-hbase.xml
+++ b/web/src/main/resources/applicationContext-hbase.xml
@@ -19,13 +19,33 @@
${hbase.client.host}
${hbase.client.port}
- ${hbase.htable.threads.max}
+
+
+ ${hbase.ipc.client.tcpnodelay}
+
+ ${hbase.rpc.timeout}
+
+ ${hbase.client.operation.timeout}
+
+
+ ${hbase.ipc.client.socket.timeout.read}
+
+ ${hbase.ipc.client.socket.timeout.write}
+
+
+
+
+
+
+
+
-
+
+
diff --git a/web/src/main/resources/hbase.properties b/web/src/main/resources/hbase.properties
index 3889f74b0..a2a44eab2 100644
--- a/web/src/main/resources/hbase.properties
+++ b/web/src/main/resources/hbase.properties
@@ -1,3 +1,22 @@
hbase.client.host=localhost
hbase.client.port=2181
-hbase.htable.threads.max=4
\ No newline at end of file
+
+# hbase timeout option==================================================================================
+# hbase default:true
+hbase.ipc.client.tcpnodelay=true
+# hbase default:60000
+hbase.rpc.timeout=10000
+# hbase default:Integer.MAX_VALUE
+hbase.client.operation.timeout=10000
+
+# hbase socket read timeout. default: 200000
+hbase.ipc.client.socket.timeout.read=20000
+# socket write timeout. hbase default: 600000
+hbase.ipc.client.socket.timeout.write=30000
+
+#==================================================================================
+# hbase client thread pool option
+hbase.client.thread.max=1024
+hbase.client.threadPool.queueSize=5120
+# prestartAllCoreThreads
+hbase.client.threadPool.prestart=false
\ No newline at end of file
diff --git a/web/src/main/resources/log4j.xml b/web/src/main/resources/log4j.xml
index 1b71ff9bc..1d8f000c1 100644
--- a/web/src/main/resources/log4j.xml
+++ b/web/src/main/resources/log4j.xml
@@ -88,8 +88,17 @@
+
+
+
+
+
+
+
+
-
+
+
\ No newline at end of file