package com.splicemachine.hbase.table;

import com.google.common.collect.Lists;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.constants.bytes.BytesUtil;
import com.splicemachine.hbase.HBaseRegionCache;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.List;
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;

public class SpliceHTableFactory implements HTableInterfaceFactory {
	private static Logger LOG = Logger.getLogger(SpliceHTableFactory.class);
	private boolean autoFlush = true;
    private ThreadPoolExecutor tableExecutor;
    private HConnection connection;

	public SpliceHTableFactory() {
        this(true);
    }
	
	public SpliceHTableFactory(boolean autoFlush) {
		SpliceLogUtils.trace(LOG, "instantiated with autoFlush set to %s",autoFlush);
		this.autoFlush = autoFlush;

        tableExecutor = getExecutor(SpliceConstants.config);
        try {
            connection = HConnectionManager.getConnection(SpliceConstants.config);
        } catch (ZooKeeperConnectionException e) {
            throw new RuntimeException(e);
        }
	}

    private ThreadPoolExecutor getExecutor(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(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Override
	  public HTableInterface createHTableInterface(Configuration config, final byte[] tableName) {
			SpliceLogUtils.trace(LOG, "createHTableInterface for %s",Bytes.toString(tableName));
	    try {
	    	final HTable htable =new SpliceHTable(tableName,connection,tableExecutor, HBaseRegionCache.getInstance());
	    	htable.setAutoFlush(autoFlush);
	    	return htable;
	    } catch (IOException ioe) {
	      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 class NamedThreadFactory implements ThreadFactory {
        private ThreadGroup group;
        private String namePrefix;
        private AtomicInteger threadNumber = new AtomicInteger(1);

        private NamedThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s!=null)? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = "htable-pool-";
        }

        @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;
        }
    }
}
@@ -48,35 +48,46 @@
 com.
-google.common.collect.Lists
+carrotsearch.hppc.IntObjectOpenHashMap
 ;%0Aim
@@ -140,60 +140,8 @@
 ts;%0A
-import com.splicemachine.constants.bytes.BytesUtil;%0A
 impo
@@ -273,24 +273,67 @@
 figuration;%0A
+import org.apache.hadoop.hbase.HConstants;%0A
 import org.a
@@ -426,121 +426,8 @@
 .*;%0A
-import org.apache.hadoop.hbase.client.coprocessor.Batch;%0Aimport org.apache.hadoop.hbase.ipc.CoprocessorProtocol;%0A
 impo
@@ -469,50 +469,8 @@
 es;%0A
-import org.apache.hadoop.hbase.util.Pair;%0A
 impo
@@ -497,17 +497,16 @@
 Logger;%0A
-%0A
 import j
@@ -529,31 +529,8 @@
 on;%0A
-import java.util.List;%0A
 impo
@@ -981,22 +981,45 @@
 ;%0A    pr
-ivate 
+otected IntObjectOpenHashMap%3C
 HConnect
@@ -1017,24 +1017,25 @@
 %3CHConnection
+%3E
  connection;
@@ -1033,17 +1033,130 @@
 nnection
-;
+s = new IntObjectOpenHashMap%3CHConnection%3E();%0A    protected static AtomicInteger increment = new AtomicInteger();  
 %0A%0A%09publi
@@ -1368,17 +1368,16 @@
 oFlush;%0A
-%0A
         
@@ -1441,26 +1441,224 @@
     
-try %7B%0A            
+%0A        for (int i = 0; i%3C 10; i++) %7B%0A        %09Configuration configuration = new Configuration(SpliceConstants.config);%0A        %09configuration.setInt(HConstants.HBASE_CLIENT_INSTANCE_ID,i);%0A        %09try %7B%0A        %09%09
 conn
@@ -1663,18 +1663,24 @@
 nnection
- =
+s.put(i,
  HConnec
@@ -1709,33 +1709,29 @@
 ion(
-SpliceConstants.config);%0A
+configuration));%0A    
     
@@ -1785,24 +1785,28 @@
             
+    
 throw new Ru
@@ -1820,24 +1820,39 @@
 ception(e);%0A
+            %7D%0A 
         %7D%0A%09%7D
@@ -2600,16 +2600,16 @@
   try %7B%0A
-
 %09    %09fi
@@ -2664,16 +2664,53 @@
 nnection
+s.get(increment.incrementAndGet()%2510)
 ,tableEx

Fix factory to have 10 connections