/*
 * Copyright 2014 NAVER Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.navercorp.pinpoint.common.hbase;

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.springframework.beans.factory.DisposableBean;

import java.io.IOException;

/**
 * HTableInterfaceFactory based on HTablePool.
 * @author emeroad
 */
public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBean {

    private HTablePool hTablePool;
    public static final int DEFAULT_POOL_SIZE = 256;

    public PooledHTableFactory(Configuration config) {
        this.hTablePool = new HTablePool(config, DEFAULT_POOL_SIZE);
    }

    public PooledHTableFactory(Configuration config, int poolSize) {
        this.hTablePool = new HTablePool(config, poolSize);
    }


    @Override
    public HTableInterface createHTableInterface(Configuration config, byte[] tableName) {
        return hTablePool.getTable(tableName);
    }

    @Override
    public void releaseHTableInterface(HTableInterface table) throws IOException {
        if (table != null) {
            table.close();
        }
    }


    @Override
    public void destroy() throws Exception {
        if (hTablePool != null) {
            this.hTablePool.close();
        }
    }
}

@@ -840,103 +840,313 @@
 nt.H
-TablePool;%0Aimport org.springframework.beans.factory.DisposableBean;%0A%0Aimport java.io.IOException
+Connection;%0Aimport org.apache.hadoop.hbase.client.HConnectionManager;%0Aimport org.springframework.beans.factory.DisposableBean;%0Aimport org.springframework.data.hadoop.hbase.HbaseSystemException;%0A%0Aimport java.io.IOException;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors
 ;%0A%0A/
@@ -1320,29 +1320,68 @@
 ate 
-HTablePool hTablePool
+ExecutorService executor;%0A    private HConnection connection
 ;%0A  
@@ -1504,359 +1504,844 @@
 his.
-hTablePool = new HTablePool(config, DEFAULT_POOL_SIZE);%0A    %7D%0A%0A    public PooledHTableFactory(Configuration config, int poolSize) %7B%0A        this.hTablePool = new HTablePool(config, poolSize);%0A    %7D%0A%0A%0A    @Override%0A    public HTableInterface createHTableInterface(Configuration config, byte%5B%5D tableName) %7B%0A        return hTablePool.getTable(tableName);
+executor = Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);%0A        try %7B%0A            this.connection = HConnectionManager.createConnection(config, executor);%0A        %7D catch (IOException e) %7B%0A            throw new HbaseSystemException(e);%0A        %7D%0A    %7D%0A%0A    public PooledHTableFactory(Configuration config, int poolSize) %7B%0A        this.executor = Executors.newFixedThreadPool(poolSize);%0A        try %7B%0A            this.connection = HConnectionManager.createConnection(config, executor);%0A        %7D catch (IOException e) %7B%0A            throw new HbaseSystemException(e);%0A        %7D%0A    %7D%0A%0A%0A    @Override%0A    public HTableInterface createHTableInterface(Configuration config, byte%5B%5D tableName) %7B%0A        try %7B%0A            return connection.getTable(tableName, executor);%0A        %7D catch (IOException e) %7B%0A            return null;%0A        %7D
 %0A   
@@ -2586,26 +2586,26 @@
     if (
-hTablePool
+connection
  != null
@@ -2629,18 +2629,18 @@
 his.
-hTablePool
+connection
 .clo

change deprecated HTablePool api to HConnection