package com.splicemachine.hbase.table;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.hbase.regioninfocache.HBaseRegionCache;
import com.splicemachine.utils.SpliceLogUtils;
import com.splicemachine.hbase.table.SpliceConnectionPool;

public class SpliceHTableFactory implements HTableInterfaceFactory {
    private static Logger LOG = Logger.getLogger(SpliceHTableFactory.class);

    private boolean autoFlush = true;
    private ExecutorService tablePool;

    private final SpliceConnectionPool connectionPool;

    public SpliceHTableFactory(){
       this(true);
    }

    public SpliceHTableFactory(boolean autoFlush){
        this(autoFlush,SpliceConnectionPool.INSTANCE);
    }

    public SpliceHTableFactory(boolean autoFlush,SpliceConnectionPool connectionPool) {
        this.connectionPool = connectionPool;
        SpliceLogUtils.trace(LOG, "instantiated with autoFlush set to %s", autoFlush);
        this.autoFlush = autoFlush;
        tablePool = createHTablePool(SpliceConstants.config);
    }

    @Override
    public HTableInterface createHTableInterface(Configuration config, final byte[] tableName) {
        SpliceLogUtils.trace(LOG, "createHTableInterface for %s", Bytes.toString(tableName));
        try {
            SpliceLogUtils.trace(LOG, "creating actual HTable after connection created");
            final HTable htable = new SpliceHTable(tableName, connectionPool.getConnectionDirect(),
                    tablePool, HBaseRegionCache.getInstance());
            htable.setAutoFlushTo(autoFlush);
            SpliceLogUtils.trace(LOG, "Returning created table");
            return htable;
        } catch (IOException ioe) {
            ioe.printStackTrace();
            throw new RuntimeException(ioe);
        }
    }

    @Override
    public void releaseHTableInterface(HTableInterface table) throws IOException {
        SpliceLogUtils.trace(LOG, "releaseHTableInterface for %s", Bytes.toString(table.getTableName()));
        table.close();
    }

    private static ExecutorService createHTablePool(Configuration config) {
        int maxThreads = config.getInt("hbase.htable.threads.max", Integer.MAX_VALUE);
        if (maxThreads == 0)
            maxThreads = 1;

        long keepAliveTime = config.getLong("hbase.htable.threads.keepalivetime", 60);

        return new ThreadPoolExecutor(1, maxThreads, keepAliveTime, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),
                new NamedThreadFactory("htable-pool-"),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    private static ExecutorService createConnectionPool(Configuration conf) {
        int coreThreads = conf.getInt("hbase.hconnection.threads.core", 10);
        if (coreThreads == 0) {
            coreThreads = 10;
        }
        int maxThreads = conf.getInt("hbase.hconnection.threads.max", 32);
        if (maxThreads == 0) {
            maxThreads = Runtime.getRuntime().availableProcessors() * 8;
        }
        long keepAliveTime = conf.getLong("hbase.hconnection.threads.keepalivetime", 60);
        LinkedBlockingQueue<Runnable> workQueue =
                new LinkedBlockingQueue<Runnable>(maxThreads *
                        conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
                                HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
        return new ThreadPoolExecutor(coreThreads,  maxThreads, keepAliveTime, TimeUnit.SECONDS,
                workQueue,
                new NamedThreadFactory("connection-pool-"),
                new ThreadPoolExecutor.AbortPolicy());

    }

    private static class NamedThreadFactory implements ThreadFactory {
        private ThreadGroup group;
        private String namePrefix;
        private AtomicInteger threadNumber = new AtomicInteger(1);

        private NamedThreadFactory(String namePrefix) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            this.namePrefix = namePrefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (!t.isDaemon())
                t.setDaemon(true);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);

            return t;
        }
    }
}
@@ -3138,1034 +3138,8 @@
  %7D%0A%0A
-    private static ExecutorService createConnectionPool(Configuration conf) %7B%0A        int coreThreads = conf.getInt(%22hbase.hconnection.threads.core%22, 10);%0A        if (coreThreads == 0) %7B%0A            coreThreads = 10;%0A        %7D%0A        int maxThreads = conf.getInt(%22hbase.hconnection.threads.max%22, 32);%0A        if (maxThreads == 0) %7B%0A            maxThreads = Runtime.getRuntime().availableProcessors() * 8;%0A        %7D%0A        long keepAliveTime = conf.getLong(%22hbase.hconnection.threads.keepalivetime%22, 60);%0A        LinkedBlockingQueue%3CRunnable%3E workQueue =%0A                new LinkedBlockingQueue%3CRunnable%3E(maxThreads *%0A                        conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,%0A                                HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));%0A        return new ThreadPoolExecutor(coreThreads,  maxThreads, keepAliveTime, TimeUnit.SECONDS,%0A                workQueue,%0A                new NamedThreadFactory(%22connection-pool-%22),%0A                new ThreadPoolExecutor.AbortPolicy());%0A%0A    %7D%0A%0A
     

Remove Unused ExecutorService