home/pinpoint
2
0
Fork 0
mirror of https://github.com/wahyd4/pinpoint.git synced 2026-08-17 00:36:02 +10:00
Code Issues Packages Projects Releases Wiki Activity

#320_backport_change_htable_api

Browse Source 
- added hbase 1.0 timeout option
 - added thread pool option
This commit is contained in:
Woonduk Kang
2015-05-28 19:43:01 +09:00
parent 2a0d3bf68d
commit dc0220cbbc
8 changed files with 159 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
collector/src/main/resources/applicationContext-hbase.xml
+20 -5
View File
@@ -19,18 +19,33 @@
<props>
<prop key="hbase.zookeeper.quorum">${hbase.client.host}</prop>
<prop key="hbase.zookeeper.property.clientPort">${hbase.client.port}</prop>
<prop key="hbase.htable.threads.max">${hbase.htable.threads.max}</prop>
<prop key="hbase.ipc.client.tcpnodelay">true</prop>
<prop key="hbase.rpc.timeout">30000</prop>
<prop key="ipc.socket.timeout">10000</prop>
<!-- hbase default:true -->
<prop key="hbase.ipc.client.tcpnodelay">${hbase.ipc.client.tcpnodelay}</prop>
<!-- hbase default:60000 -->
<prop key="hbase.rpc.timeout">${hbase.rpc.timeout}</prop>
<!-- hbase default:Integer.MAX_VALUE -->
<prop key="hbase.client.operation.timeout">${hbase.client.operation.timeout}</prop>
<!-- hbase socket read timeout. default: 200000-->
<prop key="hbase.ipc.client.socket.timeout.read">${hbase.ipc.client.socket.timeout.read}</prop>
<!-- socket write timeout. hbase default: 600000-->
<prop key="hbase.ipc.client.socket.timeout.write">${hbase.ipc.client.socket.timeout.write}</prop>
</props>
</property>
<property name="deleteConnection" value="false"/>
</bean>
<bean id="connectionFactory" class="com.navercorp.pinpoint.common.hbase.PooledHTableFactory">
<constructor-arg ref="hbaseConfiguration"/>
<constructor-arg value="${hbase.client.thread.max}"/>
<constructor-arg value="${hbase.client.threadPool.queueSize}"/>
<constructor-arg value="${hbase.client.threadPool.prestart}"/>
</bean>
<bean id="hbaseTemplate" class="com.navercorp.pinpoint.common.hbase.HbaseTemplate2">
<property name="configuration" ref="hbaseConfiguration"/>
<property name="poolSize" value="${hbase.hTablePoolSize}"/>
<property name="tableFactory" ref="connectionFactory"/>
</bean>
<bean id="applicationTraceIndexDistributor" class="com.sematext.hbase.wd.RowKeyDistributorByHashPrefix">
collector/src/main/resources/hbase.properties
+20 -1
View File
@@ -1,3 +1,22 @@
hbase.client.host=localhost
hbase.client.port=2181
hbase.htable.threads.max=4
# 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
collector/src/main/resources/pinpoint-collector.properties
-2
View File
@@ -1,5 +1,3 @@
hbase.hTablePoolSize=1024
# tcp listen ip
collector.tcpListenIp=0.0.0.0
collector.tcpListenPort=9994
commons/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java
+4 -21
View File
@@ -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();
commons/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java
+63 -10
View File
@@ -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<Runnable> discardTask = this.executor.shutdownNow();
logger.warn("discard task size:{}", discardTask.size());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}
web/src/main/resources/applicationContext-hbase.xml
+22 -2
View File
@@ -19,13 +19,33 @@
<props>
<prop key="hbase.zookeeper.quorum">${hbase.client.host}</prop>
<prop key="hbase.zookeeper.property.clientPort">${hbase.client.port}</prop>
<prop key="hbase.htable.threads.max">${hbase.htable.threads.max}</prop>
<!-- hbase default:true -->
<prop key="hbase.ipc.client.tcpnodelay">${hbase.ipc.client.tcpnodelay}</prop>
<!-- hbase default:60000 -->
<prop key="hbase.rpc.timeout">${hbase.rpc.timeout}</prop>
<!-- hbase default:Integer.MAX_VALUE -->
<prop key="hbase.client.operation.timeout">${hbase.client.operation.timeout}</prop>
<!-- hbase socket read timeout. default: 200000-->
<prop key="hbase.ipc.client.socket.timeout.read">${hbase.ipc.client.socket.timeout.read}</prop>
<!-- socket write timeout. hbase default: 600000-->
<prop key="hbase.ipc.client.socket.timeout.write">${hbase.ipc.client.socket.timeout.write}</prop>
</props>
</property>
<property name="deleteConnection" value="false"/>
</bean>
<bean id="connectionFactory" class="com.navercorp.pinpoint.common.hbase.PooledHTableFactory">
<constructor-arg ref="hbaseConfiguration"/>
<constructor-arg value="${hbase.client.thread.max}"/>
<constructor-arg value="${hbase.client.threadPool.queueSize}"/>
<constructor-arg value="${hbase.client.threadPool.prestart}"/>
</bean>
<bean id="hbaseTemplate" class="com.navercorp.pinpoint.common.hbase.HbaseTemplate2">
<property name="configuration" ref="hbaseConfiguration"></property>
<property name="configuration" ref="hbaseConfiguration"/>
<property name="tableFactory" ref="connectionFactory"/>
</bean>
<bean class="org.apache.hadoop.util.ShutdownHookManagerProxy"/>
web/src/main/resources/hbase.properties
+20 -1
View File
@@ -1,3 +1,22 @@
hbase.client.host=localhost
hbase.client.port=2181
hbase.htable.threads.max=4
# 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
web/src/main/resources/log4j.xml
+10 -1
View File
@@ -88,8 +88,17 @@
<level value="INFO"/>
<appender-ref ref="console"/>
</logger>
<logger name="org.apache.hadoop.ipc" additivity="false">
<level value="INFO"/>
<appender-ref ref="console" />
</logger>
<logger name="org.apache.hadoop" additivity="false">
<level value="INFO"/>
<appender-ref ref="console" />
</logger>
<root>
<level value="DEBUG"/>
<level value="INFO"/>
<appender-ref ref="console" />
</root>
</log4j:configuration>