@Slf4j
public class ConfigValidator { 
    public static void validate(HazeltaskConfig<?> config) {
        ExecutorConfig<?> executorConfig = config.getExecutorConfig();
        MetricsConfig metricsConfig = config.getMetricsConfig();
        
        if(config.getHazelcast() == null) {
            log.warn("No hazelcast instance provided, creating a default one to use.");
        	config.withHazelcastInstance(Hazelcast.newHazelcastInstance());
        }
        
        if(metricsConfig.getMetricsRegistry() == null) {
            metricsConfig.withMetricsRegistry(new MetricRegistry());
        }
        
        if(config.getThreadFactory() == null) {
            config.withThreadFactory(new NamedThreadFactory("Hazeltask", config.getTopologyName()));
        }
        
        if(executorConfig.getLoadBalancingConfig().getGroupPrioritizer() == null) {
            throw new IllegalArgumentException("Please specify a group prioritizer for the ExecutorConfig LoadBalancingConfig");
        }
    }
} public class ExecutorConfig<GROUP extends Serializable> {
    protected boolean          disableWorkers              = false;
    protected int              corePoolSize                = 4;
    protected int              maxPoolSize                 = 4;
    protected long             maxThreadKeepAlive          = 60000;

    @SuppressWarnings("unchecked")
    protected TaskIdAdapter<?, GROUP, ?>    taskIdAdapter     = (TaskIdAdapter<?, GROUP, ?>) new DefaultTaskIdAdapter();
    protected boolean          autoStart                   = true;
    private boolean            enableFutureTracking        = true;
    private long               maximumFutureWaitTime = TimeUnit.MINUTES.toMillis(60);
    
    private boolean            asyncronousTaskDistribution = false;
    private int                asyncronousTaskDistributionQueueSize = 500;
    private long               recoveryProcessPollInterval = 30000;
    
    private ExecutorLoadBalancingConfig<GROUP> executorLoadBalancingConfig = new ExecutorLoadBalancingConfig<GROUP>();
    private List<TaskResponseListener> taskResponseListeners = Lists.newArrayList();
    
    /**
     * Please consider using one of the ExecutorConfigs static factory methods
     */
    public ExecutorConfig() {
        
    }

    /**
     * This setting will make HazelTask buffer task distributions in a bounded queue
     * before they are sent off to the worker nodes.  This helps you get around slowness
     * in worker nodes due to GC pauses, or other load that may negatively impact 
     * Hazelcast's ability to handle calls.
     * 
     * NOTE: I think hazelcast might already actually do this???  Still testing... I will 
     * remove this code if hazelcast does do this behind the scenes.
     * 
     * @return
     */
    @Deprecated
    public ExecutorConfig<GROUP> useAsyncronousTaskDistribution() {
        asyncronousTaskDistribution = true;
        return this;
    }
    
    /**
     * When using asyncronousTaskDistribution you may define how many tasks are allowed 
     * to buffer before it start blocking the calling thread.  If a worker is JVM thrashing
     * and hazelcast is having trouble talking to it, this buffer will fill up.  This is 
     * meant to protect against JVM pauses.  You really don't want the buffer to be infinite
     * in case things go bad in your cluster.  You just want it to be big enough to get around
     * occasional JVM pauses due to GC.
     * 
     * NOTE: I think hazelcast might already actually do this???  Still testing... I will 
     * remove this code if hazelcast does do this behind the scenes.
     * 
     * @see useAsyncronousTaskDistribution()
     * @param queueSize
     * @return
     */
    @Deprecated
    public ExecutorConfig<GROUP> withAsyncronousTaskDistributionQueueSize(int queueSize) {
        this.asyncronousTaskDistributionQueueSize = queueSize;
        return this;
    }

    /**
     * This option is false by default.
     * <p>
     * If you don't want a particular member to work on tasks, enable this option
     * 
     * @param disableWorkers
     * @return
     */
    public ExecutorConfig<GROUP> withDisableWorkers(boolean disableWorkers) {
        this.disableWorkers = disableWorkers;
        return this;
    }

    /**
     * This option is not enabled by default
     * <p>
     * If you don't want a particular member to work on tasks, enable this option
     * 
     * @param disableWorkers
     * @return
     */
    public ExecutorConfig<GROUP> disableWorkers() {
        this.disableWorkers = true;
        return this;
    }
    
    /**
     * This option is not enabled by default
     * <p>
     * If you don't wish to track distributed futures you can save a little bit of work
     * by disabling them here.  This will cause this member to not push work responses 
     * into the Hazelcast response topic.
     * <p>
     * WARNING: never turn off future support on a member if other members in the cluster
     * rely on futures.  They will wait forever...
     * 
     * @return
     */
    public ExecutorConfig<GROUP> disableFutureSupport() {
        this.enableFutureTracking = false;
        return this;
    }

    public boolean isFutureSupportEnabled() {
        return this.enableFutureTracking;
    }

    /**
     * The number of threads to use to work on tasks for this member
     * 
     * @param threadCount
     * @return
     */
    public ExecutorConfig<GROUP> withThreadCount(int threadCount) {
        this.corePoolSize = threadCount;
        
        //for now we will have a fixed thread pool size
        this.maxPoolSize = threadCount;
        return this;
    }

    /**
     * This adapter is used to take a Task and identify the GROUP it belongs to
     * 
     * @param taskIdAdapter
     * @return
     */
    public ExecutorConfig<GROUP> withTaskIdAdapter(TaskIdAdapter<?,GROUP, ?> taskIdAdapter) {
        this.taskIdAdapter = taskIdAdapter;
        return this;
    }
    
    /**
     * By default the longest we will track a future is 60 minutes.  If partition data is 
     * lost, we try to figure out which futures were possible affected and error them.  This
     * setting is a fallback in case we miss erroring a future and ensures it will be cleaned up.
     * 
     * Futures waiting for longer than this time will be errored with a TimeoutException
     * 
     * @param millis
     * @return
     */
    public ExecutorConfig<GROUP> withMaximumFutureWaitTime(long millis) {
        this.maximumFutureWaitTime = millis;
        return this;
    }
    
    /**
     * This interval is the period in which the write ahead log is checked to see if any 
     * lost tasks are present.
     * 
     * @param intervalMillis
     * @return
     */
    public ExecutorConfig<GROUP> withRecoveryProcessPollInterval(long intervalMillis) {
        this.recoveryProcessPollInterval = intervalMillis;
        return this;
    }
    
    public ExecutorConfig<GROUP> addTaskResponseListener(TaskResponseListener listener) {
        this.taskResponseListeners.add(listener);
        return this;
    }
    
    public long getRecoveryProcessPollInterval() {
        return this.recoveryProcessPollInterval;
    }

    /**
     * By default we will automatically startup the task system when its
     * created. Some developers may want to delay the startup, and handle in
     * manually. For example, waiting until the web server has fully started.
     * 
     * @param autoStart
     * @return
     */
    public ExecutorConfig<GROUP> withAutoStart(boolean autoStart) {
        this.autoStart = autoStart;
        return this;
    }

    /**
     * @see withAutoStart
     */
    public ExecutorConfig<GROUP> disableAutoStart() {
        this.autoStart = false;
        return this;
    }

    public boolean isDisableWorkers() {
        return disableWorkers;
    }
   
    public int getThreadCount() {
        return corePoolSize;
    }
  
    @SuppressWarnings("unchecked")
    public TaskIdAdapter<? super Object, GROUP, ?> getTaskIdAdapter() {
        return (TaskIdAdapter<? super Object, GROUP, ?>) taskIdAdapter;
    }
 
    public boolean isAutoStart() {
        return autoStart;
    }
    
    public int getMaxThreadPoolSize() {
        return this.maxPoolSize;
    }
    
    public ExecutorConfig<GROUP> withMaxThreadKeepAliveTime(long tti) {
        this.maxThreadKeepAlive = tti;
        return this;
    }
    
    public long getMaxThreadKeepAliveTime() {
        return maxThreadKeepAlive;
    }
    
    public ExecutorConfig<GROUP> withLoadBalancingConfig(ExecutorLoadBalancingConfig<GROUP> config) {
        this.executorLoadBalancingConfig = config;
        return this;
    }
    
    public ExecutorLoadBalancingConfig<GROUP> getLoadBalancingConfig() {
        return this.executorLoadBalancingConfig;
    }

    public boolean isAsyncronousTaskDistribution() {
        return asyncronousTaskDistribution;
    }

    public int getAsyncronousTaskDistributionQueueSize() {
        return asyncronousTaskDistributionQueueSize;
    }

    public long getMaximumFutureWaitTime() {
        return maximumFutureWaitTime;
    }
    
    public List<TaskResponseListener> getTaskResponseListeners() {
        return taskResponseListeners;
    }
} public class HazeltaskConfig<GROUP extends Serializable> {
    private HazelcastInstance     hazelcast;
    private String                topologyName   = Hazeltask.DEFAULT_TOPOLOGY;
    private ExecutorConfig<GROUP> executorConfig = new ExecutorConfig<GROUP>();
    private MetricsConfig         metricsConfig  = new MetricsConfig();
    private NamedThreadFactory    threadFactory;
    
    /**
     * Topology names allow you to run multiple Hazeltask platforms on the same JVM.  You may retrive your Hazeltask intance
     * later through the Hazeltask.getInstanceByName.  The default name is defined as Hazeltask.DEFAULT_TOPOLOGY
     * 
     * @param name
     * @return
     */
    public HazeltaskConfig<GROUP> withName(String name) {
        this.topologyName = name;
        return this;
    }
    
    /**
     * Distributed executor configuration options.  The class com.hazeltask.config.defaults.ExecutorConfigs contains 
     * default common starter execution configurations
     * 
     * @see ExecutorConfigs
     * @param executorConfig
     * @return
     */
    public HazeltaskConfig<GROUP> withExecutorConfig(ExecutorConfig<GROUP> executorConfig) {
        this.executorConfig = executorConfig;
        return this;
    }
    
    /**
     * If you don't specify a hazelcast instance, we will use the default instance
     * @param hazelcast
     * @return
     */
    public HazeltaskConfig<GROUP> withHazelcastInstance(HazelcastInstance hazelcast) {
        this.hazelcast = hazelcast;
        return this;
    }
    
    /**
     * Allows the customization of Yammer metrics collected throughout Hazeltask 
     * 
     * @see http://metrics.codahale.com/
     * @param metricsConfig
     * @return
     */
    public HazeltaskConfig<GROUP> withMetricsConfig(MetricsConfig metricsConfig) {
        this.metricsConfig = metricsConfig;
        return this;
    }
    
    public MetricsConfig getMetricsConfig() {
        return this.metricsConfig;
    }
    
    public MetricRegistry getMetricsRegistry() {
        return this.metricsConfig.getMetricsRegistry();
    }

    public HazelcastInstance getHazelcast() {
        return hazelcast;
    }

    public String getTopologyName() {
        return topologyName;
    }
    
    public ExecutorConfig<GROUP> getExecutorConfig() {
        return this.executorConfig;
    }
    
    /**
     * All threads created by Hazeltask will be created through this threadFactory.  This 
     * allows you to customize all the threads used 
     * 
     * @param threadFactory
     * @return
     */
    public HazeltaskConfig<GROUP> withThreadFactory(NamedThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
        return this;
    }
    
    public NamedThreadFactory getThreadFactory() {
        return this.threadFactory;
    }
} @Slf4j
public class BackoffTimer {
    
    DelayQueue<DelayedTimerTask> queue = new DelayQueue<DelayedTimerTask>();
    private String name;
    private TimerRunnable timerRunnable;
    private final ThreadFactory threadFactory;
    private Thread workerThread;
    private volatile boolean isShutdown = false;
    
    public BackoffTimer(String name) {
        this(name, null);
    }
    
    public BackoffTimer(String name, ThreadFactory threadFactory) {
        this.name = name;
        this.threadFactory = threadFactory;
    }
    
    private void start() {
        if(isShutdown) {
            throw new IllegalStateException("BackoffTimer has been shutdown");
        }
        
        if(timerRunnable == null || !timerRunnable.newTasksMayBeScheduled) {
	    	synchronized (queue) {
	        	if(timerRunnable == null || !timerRunnable.newTasksMayBeScheduled) {
	            	timerRunnable = new TimerRunnable();
	            	if(threadFactory != null) {        		
	            		workerThread = threadFactory.newThread(timerRunnable);
	            		workerThread.start();
	            	} else {        	
	    	        	//threadFactory.newThread(r)
	    	        	//timerThread = new TimerThread(queue);
	            		workerThread = new Thread(timerRunnable);
	            		workerThread.setDaemon(false);
	            		workerThread.setName(BackoffTimer.class.getSimpleName()+"-"+name);
	            		workerThread.start();
	            	}
	            }
			}    
    	}
    }
        
    public void stop() {
    	if(timerRunnable != null) { 
	    	synchronized (queue) {
	            if(timerRunnable != null) {
	            	isShutdown = true;
	            	timerRunnable.newTasksMayBeScheduled = false;
	            	if(!queue.isEmpty()) {
	            	    Iterator<DelayedTimerTask> it = queue.iterator();
	            	    while(it.hasNext()) {
    	            	    DelayedTimerTask task = it.next();
    	            	    it.remove();
    	            	    if(task != null) {
        	            	    BackoffTask backoffTask = task.getBackoffTask();
        	            	    unschedule(backoffTask);
    	            	    }
	            	    }
	            	}
	            	queue.clear();
		            workerThread.interrupt();
		            timerRunnable = null;
		            workerThread = null;
		            queue.clear();
	            }
	        }
    	}
    }
    
    public void schedule(BackoffTask task, long minDelay, long maxDelay, double backoffMultiplier) {
    	synchronized (queue) {
	    	start();
	        queue.put(new DelayedTimerTask(task, minDelay, maxDelay, backoffMultiplier));
    	}
    }
    
    /**
     * Schedules the task with a fixed delay period and an initialDelay period.  This functions
     * like the normal java Timer.
     * @param task
     * @param initialDelay
     * @param fixedDelay
     */
    public void schedule(BackoffTask task, long initialDelay, long fixedDelay) {
    	synchronized (queue) {
	    	start();
	        queue.put(new DelayedTimerTask(task, initialDelay, fixedDelay));
    	}
    }
    
    public boolean unschedule(BackoffTask task) {
    	synchronized(queue) {
    		DelayedTimerTask wrappedTask = new DelayedTimerTask(task, 0, 0);    		
    		if(!queue.remove(wrappedTask)) {
    			//if it wasn't removed from the queue... this task may be currently running...
    			DelayedTimerTask currentTask = timerRunnable.currentTask;
    			if(currentTask != null && wrappedTask.equals(currentTask)) {
    				currentTask.forceCancel = true;
    				return true;
    			}
    		} else {
    			return true;
    		}
    		return false;
    	}
    }
    
    public static abstract class BackoffTask {
        private boolean cancelled;
        
        /**
         * 
         * @return false to backoff the next time
         */
        public abstract boolean execute();
        
        public final void cancel() {
            this.cancelled = true;
        }
        
        public final boolean isCancelled() {
            return cancelled;
        }
    }
    
    public static class DelayedTimerTask implements Delayed, Runnable {

        private final BackoffTask task;
        private long minDelay;
        private final long maxDelay;
        private final double backoffMultiplier;
        
        private long nextExecution;
        private long currentDelay;
        private boolean forceCancel = false;
        
        DelayedTimerTask(BackoffTask task, long minDelay, long maxDelay, double backoffMultiplier) {
            if(task == null)
                throw new NullPointerException("Task can't be null");
            this.task = task;
            this.minDelay = minDelay;
            this.maxDelay = maxDelay;
            this.backoffMultiplier = backoffMultiplier;
            this.currentDelay = minDelay;
        }
        
        DelayedTimerTask(BackoffTask task, long initialDelay, long fixedDelay) {
            if(task == null)
                throw new NullPointerException("Task can't be null");
            this.task = task;
            this.minDelay = fixedDelay;
            this.maxDelay = fixedDelay;
            this.backoffMultiplier = 1000;
            this.currentDelay = initialDelay;
        }
        
        public BackoffTask getBackoffTask() {
            return task;
        }
        
        public int compareTo(Delayed o) {
            return ((Long)this.getDelay(TimeUnit.MILLISECONDS)).compareTo(o.getDelay(TimeUnit.MILLISECONDS));
        }

        public long getDelay(TimeUnit unit) {
            long delayLeftMillis = nextExecution - System.currentTimeMillis();
            return unit.convert(delayLeftMillis, TimeUnit.MILLISECONDS);
        }

        public void run() {
            RuntimeException exception = null;
            //if the task throws an exception, we should backoff
            boolean needsBackoff = true;
            try {
                needsBackoff = !task.execute();
            } catch (RuntimeException e) {
                exception = e;
            }
            
            if(needsBackoff) {
                currentDelay = Math.min(maxDelay, Math.round((currentDelay * backoffMultiplier)));
            } else {
                currentDelay = minDelay;
            }
            
            nextExecution = System.currentTimeMillis() + currentDelay;
            
            if(exception != null) {
                throw exception;
            }
        }
        
        public boolean isCancelled() {
            return task.isCancelled() || forceCancel;
        }

        //delegate these methods to the internal task so that we may easily find and remove based on this
        //internal task
        
		@Override
		public int hashCode() {
			return task.hashCode();
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			DelayedTimerTask other = (DelayedTimerTask) obj;
			if (task == null) {
				if (other.task != null)
					return false;
			} else if (!task.equals(other.task))
				return false;
			return true;
		}
    }
    
    class TimerRunnable implements Runnable {
        
    	private DelayedTimerTask currentTask;
    	
    	//only access this variable in the queue monitor
    	private boolean newTasksMayBeScheduled = true;
               
        /**
         * This loop will end if the queue is empty or the thread is interrupted
         * @throws InterruptedException
         */
        public void run() {
        	while (true) {
        		try {
        			currentTask = null;
        			currentTask = queue.take();
        			if(currentTask != null) {
        				long expiredDelay = currentTask.getDelay(TimeUnit.MILLISECONDS);
        				//expired delay should optimally be 0
        				if(Math.abs(expiredDelay) >= currentTask.maxDelay / 2) {
        					//TODO: warn that the timer thread can't keep up with task execution....
        				}

        				//if the task throws an exception, it will never be run again
        				try {
        				    currentTask.run();
            				//task could have been cancelled via unschedule or the task itself
            				synchronized (queue) {
            					//put that task back so we run it again later
            					if(!currentTask.isCancelled() && newTasksMayBeScheduled)
            						queue.offer(currentTask);
            				}
        				} catch (Throwable t) {
        				    if(t instanceof InterruptedException) {
        				        throw (InterruptedException)t;
        				    }
        				    
        				    log.error("A BackoffTask: "+currentTask.getClass()+" threw an exception.  It will be cancelled. ", t);
        				    //ignore exception... just get rid of the task
        				}
        			} else {
        				//if the queue is empty, then end our thread loop
        				synchronized(queue) {
        					if(queue.isEmpty()) {
        						newTasksMayBeScheduled = false;
        						return;
        					}
        				}
        			}
        		} catch(InterruptedException e) {
        			//we were interrupted... see if its time to stop this thread
        			synchronized(queue) {
    					if(queue.isEmpty()) {
    						newTasksMayBeScheduled = false;
    						return;
    					}
    				}
        		}
        	}
        }
    }


} public interface DistributedExecutorService<GROUP extends Serializable> extends ExecutorService, ServiceListenable<DistributedExecutorService<GROUP>> {
    public void startup();
    
    @Override
    public <T> ListenableFuture<T> submit(Callable<T> task);
    
    @Override
    public ListenableFuture<?> submit(Runnable task);
    
    @Override
    public <T> ListenableFuture<T> submit(Runnable task, T result);
} @Slf4j
public class DistributedExecutorServiceImpl<GROUP extends Serializable> implements DistributedExecutorService<GROUP> {

    private ExecutorConfig<GROUP> executorConfig;
    private final HazeltaskTopology<GROUP>        topology;
    private final ListRouter<Member>       memberRouter;
    
    private final LocalTaskExecutorService<GROUP> localExecutorService;
    
    private final TaskIdAdapter<? super Object, GROUP, ?>            taskIdAdapter;
    private final DistributedFutureTracker<GROUP> futureTracker;
    private CopyOnWriteArrayList<HazeltaskServiceListener<DistributedExecutorService<GROUP>>> listeners = new CopyOnWriteArrayList<HazeltaskServiceListener<DistributedExecutorService<GROUP>>>();
    
    private final IExecutorTopologyService<GROUP>  executorTopologyService;
    
    private Timer taskAddedTimer;
    private Meter tasksRejected;
    
    //max number of times to try and submit a work before giving up
    private final int MAX_SUBMIT_TRIES = 10;
    private final int AWAIT_TERMINATION_POLL_INTERVAL = 100;
    
    private boolean isStarted = false;
    private boolean isShutdown = false;
    
    
    public DistributedExecutorServiceImpl(HazeltaskTopology<GROUP>         hcTopology, 
                                      final IExecutorTopologyService<GROUP>  executorTopologyService, 
                                      ExecutorConfig<GROUP>            executorConfig, 
                                      DistributedFutureTracker<GROUP>  futureTracker, 
                                      LocalTaskExecutorService<GROUP> localExecutorService,
                                      ExecutorMetrics metrics) {
        this.topology = hcTopology;
        this.executorConfig = executorConfig;
        this.executorTopologyService = executorTopologyService;
        
        this.memberRouter = executorConfig.getLoadBalancingConfig().getMemberRouterFactory().createRouter(new Callable<List<Member>>(){
            public List<Member> call() throws Exception {
                return topology.getReadyMembers();
            }
        });
        
        taskIdAdapter = executorConfig.getTaskIdAdapter();
        this.futureTracker = futureTracker;        
        this.localExecutorService = localExecutorService;
        
        taskAddedTimer = metrics.getTaskSubmitTimer().getMetric();
        tasksRejected = metrics.getTaskRejectedMeter().getMetric();
        
        metrics.registerLocalWriteAheadLogSizeGauge(new Gauge<Integer>(){
            @Override
            public Integer getValue() {
                return executorTopologyService.getLocalPendingTaskMapSize();
            }
        });
    }

    public ExecutorConfig<?> getExecutorConfig() {
        return this.executorConfig;
    }

    @Override
    public void execute(Runnable command) {
        Timer.Context ctx = taskAddedTimer.time();
        try {
            submitHazeltaskTask(createHazeltaskTaskWrapper(command), false);
        } finally {
            ctx.stop();
        }
    }
    
    @Override
    public void shutdown() {
        doShutdownNow(false);
    }
    
    @Override
    public List<Runnable> shutdownNow() {
        return new ArrayList<Runnable>(doShutdownNow(true));
    }
    
    public List<HazeltaskTask<GROUP>> shutdownNowWithHazeltask() {
        return doShutdownNow(true);
    }
    
    protected synchronized List<HazeltaskTask<GROUP>> doShutdownNow(boolean shutdownNow) {
        if(!isShutdown) {
            isShutdown = true;
            for(HazeltaskServiceListener<DistributedExecutorService<GROUP>> listener : listeners)
                listener.onBeginShutdown(this);
            
            //TODO: is everything shutdown?
            List<HazeltaskTask<GROUP>> tasks = null;
            if(!executorConfig.isDisableWorkers()) {
                if(shutdownNow)
                    tasks = ((LocalTaskExecutorService<GROUP>)this.localExecutorService).shutdownNow();
                else
                    this.localExecutorService.shutdown();
            }
            
            for(HazeltaskServiceListener<DistributedExecutorService<GROUP>> listener : listeners)
                listener.onEndShutdown(this);
            
            return tasks;
        }
        return Collections.emptyList();
    }

    @Override
    public boolean isShutdown() {
        return isShutdown;
    }

    @Override
    public boolean isTerminated() {
        return isShutdown && (futureTracker == null || futureTracker.size() == 0);
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
	
        if(!isShutdown)
            throw new IllegalStateException("shutdown() must be called before awaitTermination()");

        if(futureTracker == null)
            throw new IllegalStateException("FutureTracker is null");

        long elapsed = 0;
        long ms = unit.toMillis(timeout);
        long interval = Math.min(AWAIT_TERMINATION_POLL_INTERVAL, ms);

        do {
            if(futureTracker.size() == 0) 
                return true;

            Thread.sleep(interval);
            elapsed += interval;

        } while(elapsed < ms);

        return false;
    }

    @Override
    public <T> ListenableFuture<T> submit(Callable<T> task) {
        Timer.Context ctx = taskAddedTimer.time();
        try {
            if(futureTracker == null)
                throw new IllegalStateException("FutureTracker is null");
            
            HazeltaskTask<GROUP> taskWrapper = createHazeltaskTaskWrapper(task);
            DistributedFuture<GROUP, T> future = futureTracker.createFuture(taskWrapper);
            if(!submitHazeltaskTask(taskWrapper, false)) {
                //remove future from tracker, error out future with duplicate exception
                //i hate this... it would be a cool feature to attach this future to the 
                //work in progress.  its easier to just cancel it for now
                //TODO: should we cancel or throw an exception?
                //  - i think cancel, because presumably the work is going to run we just can't track it
                //    so if you don't care about the result, no harm
                log.error("Unable to submit HazeltaskTask to worker member");
                future.setCancelled(false);
                futureTracker.remove(taskWrapper.getId());
            }
            return future;
        } finally {
            ctx.stop();
        }
    }
    
    private void validateTask(Object task) {
        if(!(task instanceof Serializable)) {
            throw new IllegalArgumentException("The task type "+task.getClass()+" must implement Serializable");
        }
        
        if(!taskIdAdapter.supports(task)) {
            throw new IllegalArgumentException("This executor doesn't support the type "+task.getClass());
        }
    }
    
    @SuppressWarnings("unchecked")
    private HazeltaskTask<GROUP> createHazeltaskTaskWrapper(Runnable task){
        if(task instanceof HazeltaskTask) {
            ((HazeltaskTask<GROUP>) task).updateCreatedTime();
            return (HazeltaskTask<GROUP>) task;
        } else {
            validateTask(task);            
            return new HazeltaskTask<GROUP>(UUID.randomUUID(), 
                                     taskIdAdapter.getTaskGroup(task), 
                                     taskIdAdapter.getTaskInfo(task),
                                     task);
        }
    }
    
    private HazeltaskTask<GROUP> createHazeltaskTaskWrapper(Callable<?> task) {
        validateTask(task); 
        return new HazeltaskTask<GROUP>(UUID.randomUUID(), 
                                 taskIdAdapter.getTaskGroup(task), 
                                 taskIdAdapter.getTaskGroup(task),
                                 task);
    }

    @Override
    public <T> ListenableFuture<T> submit(Runnable task, T result) {
        Timer.Context ctx = taskAddedTimer.time();
        try {
            throw new RuntimeException("Not Implemented Yet");
        } finally {
            ctx.stop();
        }   
    }

    @Override
    public ListenableFuture<?> submit(Runnable task) {
        Timer.Context ctx = taskAddedTimer.time();
        try {
            if(futureTracker == null)
                throw new IllegalStateException("FutureTracker is null");
            
            HazeltaskTask<GROUP> taskWrapper = createHazeltaskTaskWrapper(task);
            DistributedFuture<GROUP, ?> future = futureTracker.createFuture(taskWrapper);
            submitHazeltaskTask(taskWrapper, false);
            return future;
        } finally {
            ctx.stop();
        }
    }
    
    public boolean submitHazeltaskTask(HazeltaskTask<GROUP> wrapper, boolean isResubmitting) {      
        
        //WorkId workKey = wrapper.getWorkId();
        boolean executeTask = true;
        /*
         * with acknowledgeWorkSubmition, we will sit in this loop until a 
         * node accepts our work item.  Currently, a node will accept as long as
         * a MemberLeftException is not thrown
         */
        int tries = 0;
        while(++tries <= MAX_SUBMIT_TRIES) {
            if(isResubmitting) {
                wrapper.setSubmissionCount(wrapper.getSubmissionCount()+1);
                executorTopologyService.addPendingTask(wrapper, true);
            } else {
                executeTask = executorTopologyService.addPendingTask(wrapper, false);
            }
            
            if(executeTask) {
                Member m = memberRouter.next();
                if(m == null) {
                    log.warn("Work submitted to writeAheadLog but no members are online to do the work.");
                    tasksRejected.mark();
                    return false;
                }
                
                try {
                    executorTopologyService.sendTask(wrapper, m);
                    return true;
                } catch (RuntimeException e) {
                    log.error("Tried to distribute task, but I got an exception",e);
                    isResubmitting = true;
                } catch (TimeoutException e) {
                    log.warn("Timed out while trying to submit task for try #"+tries+", trying again...");
                    isResubmitting = true;
                }
            } else {
                //do not submit
                tasksRejected.mark();
                return false;
            }
        }
        
        tasksRejected.mark();
        throw new RuntimeException("Unable to submit work to nodes. I tried "+MAX_SUBMIT_TRIES+" times.");
    }

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
        // TODO Auto-generated method stub
        throw new RuntimeException("Not Implemented Yet");
    }

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout,
            TimeUnit unit) throws InterruptedException {
        // TODO Auto-generated method stub
        throw new RuntimeException("Not Implemented Yet");
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException,
            ExecutionException {
        // TODO Auto-generated method stub
        throw new RuntimeException("Not Implemented Yet");
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
        // TODO Auto-generated method stub
        throw new RuntimeException("Not Implemented Yet");
    }

    @Override
    public synchronized void startup() {
        if(!isStarted) {
            for(HazeltaskServiceListener<DistributedExecutorService<GROUP>> listener : listeners)
                listener.onBeginStart(this);
            //TODO: clean up this code.  There isn't really a startup process anymore
            for(HazeltaskServiceListener<DistributedExecutorService<GROUP>> listener : listeners)
                listener.onEndStart(this);
            
            isStarted = true;
        }
    }

    @Override
    public void addServiceListener(HazeltaskServiceListener<DistributedExecutorService<GROUP>> listener) {
        this.listeners.add(listener);
    }
    
    
    public void addLocalExecutorListener(ExecutorListener<GROUP> listener) {
        if(!executorConfig.isDisableWorkers())
            this.localExecutorService.addListener(listener);
    }
    
    public LocalTaskExecutorService<GROUP> getLocalTaskExecutorService() {
        return (LocalTaskExecutorService<GROUP>) this.localExecutorService;
    }

} @Slf4j
public class DistributedFutureTracker<GROUP extends Serializable> implements MessageListener<TaskResponse<Serializable>> {
    private Cache<UUID, DistributedFuture<GROUP, Serializable>> futures;
    
    private final Histogram futureWaitTimeHistogram;
    private final IExecutorTopologyService<GROUP> topologyService;
    
    /**
     * 
     * @param metrics (nullable)
     */
    public DistributedFutureTracker(final IExecutorTopologyService<GROUP> topologyService, ExecutorMetrics metrics, ExecutorConfig<GROUP> config) {
        this.topologyService = topologyService;
        futures = CacheBuilder.newBuilder()
                //no future will wait for more than this time
                .expireAfterAccess(config.getMaximumFutureWaitTime(), TimeUnit.MILLISECONDS)
                .removalListener(new RemovalListener<UUID, DistributedFuture<GROUP, Serializable>>() {
                    @Override
                    public void onRemoval(RemovalNotification<UUID, DistributedFuture<GROUP, Serializable>> notification) {
                        if(notification.getCause() == RemovalCause.EXPIRED) {
                        	DistributedFuture<GROUP, Serializable> future = notification.getValue();
                            long waitTimeMillis = System.currentTimeMillis() - future.getCreatedTime();
                            notification.getValue().setException(new TimeoutException("Future timed out waiting.  Waited "+(TimeUnit.MILLISECONDS.toMinutes(waitTimeMillis))+" minutes"));
                            
                            topologyService.cancelTask(future.getGroup(), future.getTaskId()); 
                            topologyService.removePendingTask(future.getTaskId());
                        } else if(notification.getCause() == RemovalCause.COLLECTED) {
                            //future was GC'd because we didn't want to track it
                            log.debug("Future "+notification.getKey()+" was garabge collected and removed from the tracker");
                        }
                    }
                 })
                .build();
        
        if(metrics != null) {
            metrics.registerLocalFuturesWaitingGauge(new LocalFuturesWaitingGauge(this));
            futureWaitTimeHistogram = metrics.getFutureWaitTimeHistogram().getMetric();
        } else {
            futureWaitTimeHistogram = null;
        }
    }
    
    //It is required that T be Serializable
    @SuppressWarnings("unchecked")
    public <T> DistributedFuture<GROUP, T> createFuture(HazeltaskTask<GROUP> task) {
        DistributedFuture<GROUP, T> future = new DistributedFuture<GROUP, T>(topologyService, task.getGroup(), task.getId());
        this.futures.put(task.getId(), (DistributedFuture<GROUP, Serializable>) future);
        return future;
    }
    
    protected DistributedFuture<GROUP, Serializable> remove(UUID id) {
        DistributedFuture<GROUP, Serializable> f =futures.getIfPresent(id); 
        futures.invalidate(id);
        
       if(f != null) {
         if(futureWaitTimeHistogram != null) {
            futureWaitTimeHistogram.update(System.currentTimeMillis() - f.getCreatedTime());
         }
       }
         
         return f;
    }
    
    public Set<UUID> getTrackedTaskIds() {
        return new HashSet<UUID>(futures.asMap().keySet());
    }
    
    @Override
    public void onMessage(Message<TaskResponse<Serializable>> message) {
        TaskResponse<Serializable> response = message.getMessageObject();
        UUID taskId = response.getTaskId();
        DistributedFuture<GROUP, Serializable> future = remove(taskId);
        if(future != null) {
            if(response.getStatus() == Status.FAILURE) {
                future.setException(response.getError());
            } else if(response.getStatus() == Status.SUCCESS) {
                future.set((Serializable)response.getResponse());
            } else if (response.getStatus() == Status.CANCELLED) {
                //TODO: add a status for INTERRUPTED
                future.setCancelled(false);
            }
            
        }
    }

    /**
     * handles when a member leaves and hazelcast partition data is lost.  We want 
     * to find the Futures that are waiting on lost data and error them
     */
    public void errorFuture(UUID taskId, Exception e) {
        DistributedFuture<GROUP, Serializable> future = remove(taskId);
        if(future != null) {
            future.setException(e);
        }
    }
    
    public int size() {
    	return (int) futures.size();
    }
    
    
} @Slf4j
public class HazelcastExecutorTopologyService<GROUP extends Serializable> implements IExecutorTopologyService<GROUP> {
    //private final BloomFilter<CharSequence> bloomFilter;
    private HazeltaskTopology<GROUP> topology;
    private String topologyName;
    private final Member me;
    
    private final IExecutorService communicationExecutorService;

    private final IExecutorService taskDistributor;
    //private final CopyOnWriteArrayListSet<Member> readyMembers;
    private final IMap<UUID, HazeltaskTask<GROUP>>                            pendingTask;
    private final ILock rebalanceTasksLock;
    private final ITopic<TaskResponse<Serializable>>      taskResponseTopic;
    private final HazelcastInstance hazelcast;
    
    private final Executor asyncTaskDistributorExecutor;
    
    public HazelcastExecutorTopologyService(HazeltaskConfig<GROUP> hazeltaskConfig, HazeltaskTopology<GROUP> topology) {
        com.hazeltask.config.ExecutorConfig<GROUP> executorConfig = hazeltaskConfig.getExecutorConfig();
        topologyName = hazeltaskConfig.getTopologyName();
        this.topology = topology;
        hazelcast = hazeltaskConfig.getHazelcast();
        this.me = hazelcast.getCluster().getLocalMember();
        communicationExecutorService = hazelcast.getExecutorService(name("com"));
        
        String taskDistributorName = name("task-distributor");
        
        //limit the threads on the distributor to 1 thread
        hazelcast.getConfig()
            .addExecutorConfig(new ExecutorConfig()
                .setName(taskDistributorName)
                .setPoolSize(1)
            );
        
        taskDistributor =  hazelcast.getExecutorService(taskDistributorName);
        //readyMembers = new CopyOnWriteArrayListSet<Member>();
        
        if(hazeltaskConfig.getExecutorConfig().isAsyncronousTaskDistribution())
            asyncTaskDistributorExecutor =  new ThreadPoolExecutor(1, 1,
                            0L, TimeUnit.MILLISECONDS,
                            new LinkedBlockingQueue<Runnable>(executorConfig.getAsyncronousTaskDistributionQueueSize()),
                            hazeltaskConfig.getThreadFactory().named("async"),
                            new CallerRunsPolicy());
        else
            asyncTaskDistributorExecutor = null;
        
        String pendingTaskMapName = name("pending-tasks");
        hazelcast.getConfig()
        .addMapConfig(new MapConfig()
            .setName(taskDistributorName)
            .addMapIndexConfig(new MapIndexConfig("createdAtMillis", false)));
        
        pendingTask = hazelcast.getMap(pendingTaskMapName);
        taskResponseTopic = hazelcast.getTopic(name("task-response"));
        
        rebalanceTasksLock = hazelcast.getLock(name("task-balance"));
    }
    
    private String name(String name) {
        return topologyName + "-" + name;
    }
    
    public void sendTask(HazeltaskTask<GROUP> task, Member member) throws TimeoutException {
    	SubmitTaskOp<GROUP> distTask = new SubmitTaskOp<GROUP>(task, topologyName);
        if(asyncTaskDistributorExecutor != null) {
            asyncTaskDistributorExecutor.execute(new $SendTaskToWorker(taskDistributor, member, distTask));
        } else {
            taskDistributor.submitToMember(distTask, member);
        }
    }
    
    @RequiredArgsConstructor
    private static class $SendTaskToWorker implements Runnable {
        private final IExecutorService taskDistributor;
        private final Member member;
        private final Callable<Boolean> task;
        
        @Override
        public void run() {
            taskDistributor.submitToMember(task, member);
        }
    }

    /**
     * Add to the write ahead log (hazelcast IMap) that tracks all the outstanding tasks
     */
    public boolean addPendingTask(HazeltaskTask<GROUP> task, boolean replaceIfExists) {
        if(!replaceIfExists)
            return pendingTask.putIfAbsent(task.getId(), task) == null;
        
        pendingTask.put(task.getId(), task);
        return true;
    }
    
    /**
     * Asynchronously put the work into the pending map so we can work on submitting it to the worker
     * if we wanted.  Could possibly cause duplicate work if we execute the work, then add to the map.
     * @param task
     * @return
     */
    public Future<HazeltaskTask<GROUP>> addPendingTaskAsync(HazeltaskTask<GROUP> task) {
        return pendingTask.putAsync(task.getId(), task);
    }

    public boolean removePendingTask(HazeltaskTask<GROUP> task) {
        return removePendingTask(task.getId());
    }
    
    public boolean removePendingTask(UUID taskId) {
    	pendingTask.removeAsync(taskId);
    	return true;
    }

    public void broadcastTaskCompletion(UUID taskId, Serializable response, Serializable taskInfo) {
        TaskResponse<Serializable> message = new TaskResponse<Serializable>(me, taskId, taskInfo, response, TaskResponse.Status.SUCCESS);
        taskResponseTopic.publish(message);
    }

    public void broadcastTaskCancellation(UUID taskId, Serializable taskInfo) {
        TaskResponse<Serializable> message = new TaskResponse<Serializable>(me, taskId, taskInfo, null, TaskResponse.Status.CANCELLED);
        taskResponseTopic.publish(message);
    }

    public void broadcastTaskError(UUID taskId, Throwable exception, Serializable taskInfo) {
        TaskResponse<Serializable> message = new TaskResponse<Serializable>(me, taskId, taskInfo, exception);
        taskResponseTopic.publish(message);
    }

    public Collection<HazeltaskTask<GROUP>> getLocalPendingTasks(String predicate) {
        Set<UUID> keys = pendingTask.localKeySet(new SqlPredicate(predicate));
        return pendingTask.getAll(keys).values();
    }

    public Collection<MemberResponse<Long>> getMemberQueueSizes() {
        return MemberTasks.executeOptimistic(
                communicationExecutorService, 
                topology.getReadyMembers(),
                new GetLocalQueueSizesOp<GROUP>(topology.getName())
        );
    }
    


    @Override
    public Collection<MemberResponse<Map<GROUP, Integer>>> getMemberGroupSizes() {
        return MemberTasks.executeOptimistic(
                communicationExecutorService, 
                topology.getReadyMembers(),
                new GetLocalGroupQueueSizesOp<GROUP>(topology.getName())
        );
    }

    public void addTaskResponseMessageHandler(MessageListener<TaskResponse<Serializable>> listener) {
        taskResponseTopic.addMessageListener(listener);
    }

    

    

    public Lock getRebalanceTaskClusterLock() {
        return rebalanceTasksLock;
    }

    public Collection<HazeltaskTask<GROUP>> stealTasks(List<MemberValuePair<Long>> numToTake) {
        Collection<HazeltaskTask<GROUP>> result = new LinkedList<HazeltaskTask<GROUP>>();
        Collection<Future<Collection<HazeltaskTask<GROUP>>>> futures = new ArrayList<Future<Collection<HazeltaskTask<GROUP>>>>(numToTake.size());
        for(MemberValuePair<Long> entry : numToTake) {    	
        	futures.add(communicationExecutorService
        					.submitToMember(
        							new StealTasksOp<GROUP>(topology.getName(), entry.getValue()), 
        							entry.getMember()
        					)
        				);
        }
        
        for(Future<Collection<HazeltaskTask<GROUP>>> f : futures) {
            try {
                Collection<HazeltaskTask<GROUP>> task = f.get(3, TimeUnit.MINUTES);//wait at most 3 minutes
                result.addAll(task);
            } catch (InterruptedException e) {
                log.error("Unable to take tasks. I was interrupted.  We may have pulled work out of another member... it will need to be recovered", e);
                Thread.currentThread().interrupt();
                return result;
            } catch (ExecutionException e) {
                log.error("Unable to take tasks. I got an exception.  We may have pulled work out of another member... it will need to be recovered", e);
                continue;
            } catch (TimeoutException e) {
                log.error("Unable to take tasks within 3 minutes.  We may have pulled work out of another member... it will need to be recovered");
                continue;
            } 
        }
        return result;
    }

    public int getLocalPendingTaskMapSize() {
        return pendingTask.localKeySet().size();
    }

    public Collection<MemberResponse<Long>> getOldestTaskTimestamps() {
        return MemberTasks.executeOptimistic(
             communicationExecutorService, 
             topology.getReadyMembers(),
             new GetOldestTimestampOp<GROUP>(topology.getName())
        );
    }
    
    @Override
    public Collection<MemberResponse<Integer>> getThreadPoolSizes() {
        return MemberTasks.executeOptimistic(
                communicationExecutorService, 
                topology.getReadyMembers(),
                new GetThreadPoolSizesOp<GROUP>(topology.getName())
        );
    }

    @Override
    public Collection<MemberResponse<Map<GROUP, Integer>>> getGroupSizes(Predicate<GROUP> predicate) {
        if(predicate != null && !(predicate instanceof Serializable)) {
            //using illegalargument instead of notserializable because its dumb to have a checked exception here
            throw new IllegalArgumentException(predicate.getClass().getName()+" is not serializable");
        }
        
        return MemberTasks.executeOptimistic(
                communicationExecutorService, 
                topology.getReadyMembers(),
                new GetLocalGroupQueueSizesOp<GROUP>(topology.getName(), predicate)
        );
    }

    @Override
    public void clearGroupQueue(GROUP group) {
        MemberTasks.executeOptimistic(
                communicationExecutorService, 
                topology.getReadyMembers(),
                new ClearGroupQueueOp<GROUP>(topology.getName(), group)
        );
    }

    @Override
    public boolean cancelTask(GROUP group, UUID taskId) {
        Collection<MemberResponse<Boolean>> responses = MemberTasks.executeOptimistic(
             communicationExecutorService, 
             topology.getReadyMembers(),
             new CancelTaskOp<GROUP>(topology.getName(), taskId, group)
        );
        
        for(MemberResponse<Boolean> response : responses) {
            if(response.getValue() == true)
                return true;
        }
        return false;
    }
} public interface IExecutorTopologyService<GROUP extends Serializable> {
    //public boolean isMemberReady(Member member);
    
    public void sendTask(HazeltaskTask<GROUP> task, Member member) throws TimeoutException;
    
    
    /**
     * 
     * @param task
     * @param replaceIfExists
     * @return
     */
    public boolean addPendingTask(HazeltaskTask<GROUP> task, boolean replaceIfExists);
    
    /**
     * Retrive the hazeltasks in the local pending task map with the predicate restriction
     * @param predicate
     * @return
     */
    public Collection<HazeltaskTask<GROUP>> getLocalPendingTasks(String predicate);
    
    /**
     * Get the local queue sizes for each member
     * 
     * 
     * @return
     */
    public Collection<MemberResponse<Long>> getMemberQueueSizes();
    
    /**
     * Get the local queue sizes for each group on each member
     * 
     * 
     * @return
     */
    public Collection<MemberResponse<Map<GROUP, Integer>>> getMemberGroupSizes();
    
    /**
     * Get the local partition's size of the pending work map
     * TODO: should this live in a different Service class?
     * @return
     */
    public int getLocalPendingTaskMapSize();
    
    public Collection<MemberResponse<Long>> getOldestTaskTimestamps();
    
    /**
     * 
     * @param task
     * @return true if removed, false it did not exist
     */
    public boolean removePendingTask(HazeltaskTask<GROUP> task);
    public boolean removePendingTask(UUID taskId);
    
    public void broadcastTaskCompletion(UUID taskId, Serializable response, Serializable taskInfo);
    public void broadcastTaskCancellation(UUID taskId, Serializable taskInfo);
    public void broadcastTaskError(UUID taskId, Throwable exception, Serializable taskInfo);
    public void addTaskResponseMessageHandler(MessageListener<TaskResponse<Serializable>> listener);
    
    public Lock getRebalanceTaskClusterLock();
    
    public Collection<HazeltaskTask<GROUP>> stealTasks(List<MemberValuePair<Long>> numToTake);
    //public boolean addTaskToLocalQueue(HazelcastWork task);
    
    public Collection<MemberResponse<Integer>> getThreadPoolSizes();
    public Collection<MemberResponse<Map<GROUP, Integer>>> getGroupSizes(Predicate<GROUP> predicate);
    public void clearGroupQueue(GROUP group);
    
    public boolean cancelTask(GROUP group, UUID taskId);
} public interface TaskResponseListener extends MessageListener<TaskResponse<Serializable>> {
    
} @Slf4j
public class LocalTaskExecutorService<G extends Serializable> {

	private final HazeltaskThreadPoolExecutor localExecutorPool;
	private final GroupedPriorityQueueLocking<HazeltaskTask<G>, G> taskQueue;
	private final TasksInProgressTracker tasksInProgressTracker;
	private final HazelcastInstance hazelcast;
	private final IExecutorTopologyService<G> executorTopologyService;
	private final ExecutorConfig<G> executorConfig;
	
	private final Timer taskSubmittedTimer;
	private final Timer taskExecutedTimer;
	
	private final Timer getGroupSizesTimer;
    private final Timer getOldestTaskTimeTimer;
    private final Timer getQueueSizeTimer;
    
    private final Meter taskErrorsMeter;
    private final Timer removeFromWriteAheadLogTimer;
    private final Timer taskFinishedNotificationTimer;
	
    public LocalTaskExecutorService(HazelcastInstance hazelcast, ExecutorConfig<G> executorConfig, NamedThreadFactory namedThreadFactory, IExecutorTopologyService<G> executorTopologyService, ExecutorMetrics metrics) {
		this.hazelcast = hazelcast;
		
		taskQueue = new GroupedPriorityQueueLocking<HazeltaskTask<G>, G>(metrics, executorConfig.getLoadBalancingConfig().getGroupPrioritizer());

		taskSubmittedTimer = metrics.getLocalTaskSubmitTimer().getMetric();
		taskExecutedTimer = metrics.getTaskExecutionTimer().getMetric();
		getGroupSizesTimer = metrics.getGetGroupSizesTimer().getMetric();
		getOldestTaskTimeTimer = metrics.getGetOldestTaskTimeTimer().getMetric();
		getQueueSizeTimer = metrics.getGetQueueSizeTimer().getMetric();
		taskErrorsMeter = metrics.getTaskErrors().getMetric();
		removeFromWriteAheadLogTimer = metrics.getRemoveFromWriteAheadLogTimer().getMetric();
		taskFinishedNotificationTimer = metrics.getTaskFinishedNotificationTimer().getMetric();
		
		metrics.registerCollectionSizeGauge(new CollectionSizeGauge(taskQueue));
		metrics.registerExecutionThroughputGauge(new TaskThroughputGauge(taskSubmittedTimer, taskExecutedTimer));
		
		@SuppressWarnings({ "rawtypes", "unchecked" })
        BlockingQueue<Runnable> blockingQueue = (BlockingQueue<Runnable>) (BlockingQueue) taskQueue;
		
		localExecutorPool = new HazeltaskThreadPoolExecutor(
		        executorConfig.getThreadCount(), 
		        executorConfig.getMaxThreadPoolSize(), 
		        executorConfig.getMaxThreadKeepAliveTime(), 
		        TimeUnit.MILLISECONDS, 
		        blockingQueue, 
		        namedThreadFactory.named("worker"), 
		        new AbortPolicy());
		
		if(executorConfig.isFutureSupportEnabled())
		    localExecutorPool.addListener(new ResponseExecutorListener<G>(executorTopologyService, taskFinishedNotificationTimer));
		
		localExecutorPool.addListener(new TaskCompletionExecutorListener<G>(executorTopologyService, taskErrorsMeter, removeFromWriteAheadLogTimer));
		
		tasksInProgressTracker = new TasksInProgressTracker();
		localExecutorPool.addListener(tasksInProgressTracker);
		
		this.executorTopologyService = executorTopologyService;
		this.executorConfig = executorConfig;
	}
	
	/**
     * This is not thread safe
     * @param listener
     */
    public void addListener(ExecutorListener<G> listener) {
        localExecutorPool.addListener(listener);
    }
    
    private class TasksInProgressTracker implements ExecutorListener<G> {
        private final Map<UUID, HazeltaskTask<G>> tasksInProgress = new ConcurrentHashMap<UUID, HazeltaskTask<G>>();

        @Override
        public void beforeExecute(HazeltaskTask<G> runnable) {
            tasksInProgress.put(runnable.getId(), runnable);
        }

        @Override
        public void afterExecute(HazeltaskTask<G> runnable, Throwable exception) {
            tasksInProgress.remove(runnable.getId());
        }
        
        public long getOldestTime() {
            long oldestTime = Long.MAX_VALUE;
            for(HazeltaskTask<G> task : tasksInProgress.values()) {
                if(task.getTimeCreated() < oldestTime) {
                    oldestTime = task.getTimeCreated();
                }
            }
            return oldestTime;
        }
        
//        public boolean cancelTask(UUID uuid) {
//            HazeltaskTask<G> task = tasksInProgress.get(uuid);
//            if(task != null) {
//                //FIXME: we need the reference to the future to do the interrupt
//                return true;
//            }
//            return false;
//        }
        
    }
    
    private static class TaskCompletionExecutorListener< G extends Serializable> implements ExecutorListener<G> {
        private final IExecutorTopologyService<G> executorTopologyService;
        private final Meter taskErrorsMeter;
        private final Timer removeFromWriteAheadLogTimer;
        
        public TaskCompletionExecutorListener(IExecutorTopologyService<G> executorTopologyService, Meter taskErrorsMeter, Timer removeFromWriteAheadLogTimer) {
            this.executorTopologyService = executorTopologyService;
            this.taskErrorsMeter = taskErrorsMeter;
            this.removeFromWriteAheadLogTimer = removeFromWriteAheadLogTimer;
        }
        
        public void afterExecute(HazeltaskTask<G> runnable, Throwable exception) {
            if(exception != null) {
                taskErrorsMeter.mark();
            }
            
            HazeltaskTask<G> task = (HazeltaskTask<G>)runnable;
            //TODO: add task exceptions handling / retry logic
            //for now, just remove the work because its completed
            Timer.Context ctx = removeFromWriteAheadLogTimer.time();
            try {
                executorTopologyService.removePendingTask(task);
            } finally {
                ctx.stop();
            }
        }

        public void beforeExecute(HazeltaskTask<G> runnable) {}
    }
	
	/**
	 * We want to get the best result possible as this value 
	 * is used to determine what work needs to be recovered.
	 * 
	 * @return
	 */
	public Long getOldestTaskCreatedTime(){
	    Timer.Context ctx = getOldestTaskTimeTimer.time();
	    try {
    	    long oldest = Long.MAX_VALUE;
    	    
    	    /*
    	     * I am asking this question first, because if I ask it after I could
    	     * miss the oldest time if the oldest is polled and worked on
    	     */
    	    Long oldestQueueTime = this.taskQueue.getOldestQueueTime();
    	    if(oldestQueueTime != null)
    	        oldest = oldestQueueTime;
    	    
    	    //there is a tiny race condition here... but we just want to make our best attempt
    	    long inProgressOldestTime = tasksInProgressTracker.getOldestTime();
    	    
    	    if(inProgressOldestTime < oldest)
    	        oldest = inProgressOldestTime;
    	    
    	    return oldest;
	    } finally {
	        ctx.stop();
	    }
	}
	
	public long getQueueSize() {
	    Timer.Context ctx = getQueueSizeTimer.time();
	    try {
	        return this.taskQueue.size();
	    } finally {
	        ctx.stop();
	    }
	}
	
	public Map<G, Integer> getGroupSizes() {
	    return getGroupSizes(null);
	}
	
	/**
	 * TODO: should we index this predicate somehow?
	 * 
	 * @param predicate
	 * @return
	 */
	public Map<G, Integer> getGroupSizes(Predicate<G> predicate) {
        Timer.Context ctx = getGroupSizesTimer.time();
        try {
            return this.taskQueue.getGroupSizes(predicate);
        } finally {
            ctx.stop();
        }
    }
    
    public void clearGroup(G group) {
        Queue<HazeltaskTask<G>> q = taskQueue.getQueueByGroup(group);
        Iterator<HazeltaskTask<G>> queueIterator = q.iterator();
        while(queueIterator.hasNext()) {
            try {
                HazeltaskTask<G> next = queueIterator.next();
                if(executorConfig.isFutureSupportEnabled())
                    executorTopologyService.broadcastTaskCancellation(next.getId(), next.getTaskInfo());
                executorTopologyService.removePendingTask(next);
                queueIterator.remove();            
            } catch (NoSuchElementException e) {
                return;//bail out
            }
        }
    }
	
	public void execute(HazeltaskTask<G> command) {
		if(localExecutorPool.isShutdown()) {
		    log.warn("Cannot enqueue the task "+command+".  The executor threads are shutdown.");
		    return;
		}
	    
	    Timer.Context tCtx = null;
		if(taskSubmittedTimer != null)
			tCtx = taskSubmittedTimer.time();
		try {
			command.setExecutionTimer(taskExecutedTimer);
		    command.setHazelcastInstance(hazelcast);
			localExecutorPool.execute(command);
		} finally {
			if(tCtx != null)
				tCtx.stop();
		}
	}
	
	public Collection<HazeltaskTask<G>> stealTasks(long numberOfTasks) {
	    if(!this.localExecutorPool.isShutdown()) {
    	    long totalSize = taskQueue.size();
    	    ArrayList<HazeltaskTask<G>> result = new ArrayList<HazeltaskTask<G>>((int)numberOfTasks);
    	    for(G group : this.taskQueue.getGroups()) {
    	        ITrackedQueue<HazeltaskTask<G>> q = this.taskQueue.getQueueByGroup(group);
    	        int qSize = q.size();
    	        if(qSize == 0) continue;
    	        
    	        double p = (double)qSize / (double)totalSize;
    	        long tasksToTake = Math.round(numberOfTasks * p);
    	        
    	        for(int i=0; i < tasksToTake; i++) {
    	            //TODO: this really sucks that we use q.poll() ... why can't this be a dequeue????
    	            HazeltaskTask<G> task = q.poll();
    	            if(task == null)
    	                break;
    	            result.add(task);
    	        }
    	    }
    	    
    	    if(result.size() < numberOfTasks) {
    	        //TODO: should we really care? or is this good enough...
    	    }   
    	    
    	    return result;
	    } else {
	        log.warn("Cannot steal "+numberOfTasks+" tasks.  The executor threads are shutdown.");
	        return Collections.emptyList();
	    }
	}

	//TODO: time how long it takes to shutdown
	public void shutdown() {
	    localExecutorPool.shutdown();
	}
	
	//TODO: time how long it takes to shutdown
	//SuppressWarnings I really want to return HazeltaskTasks instead of Runnable
	@SuppressWarnings({ "unchecked", "rawtypes" })
    public List<HazeltaskTask<G>> shutdownNow() {
	    return (List<HazeltaskTask<G>>) (List) localExecutorPool.shutdownNow();
	}

	public boolean isShutdown() {
		return localExecutorPool.isShutdown();
	}

    public Boolean cancelTask(UUID taskId, G group) {
        ITrackedQueue<HazeltaskTask<G>> queue = this.taskQueue.getQueueByGroup(group);
        if(queue != null) {
            Iterator<HazeltaskTask<G>> it = queue.iterator();
            while(it.hasNext()) {
                HazeltaskTask<G> task = it.next();
                if(task.getId().equals(taskId)) {
                    if(executorConfig.isFutureSupportEnabled())
                        executorTopologyService.broadcastTaskCancellation(taskId, task.getTaskInfo());                
                    it.remove();
                    return true;
                }
            }
        }
        
        //TODO: allow cancelling of inprogress tasks but we need access to the Thread that is running it
        return false;
    }

} public class ExecutorMetrics {
	private final String topologyName;
	private final MetricRegistry metrics;

	private final Metric<Timer> staleTaskFlushTimer;
	private final Metric<Histogram> staleFlushCountHistogram;

	private final Metric<Timer> taskBalanceTimer;
	private final Metric<Histogram> taskBalanceHistogram;
	private final Metric<Timer> taskBalanceLockWaitTimer;

	private final Metric<Timer> taskSubmitTimer;
	private final Metric<Meter> taskRejectedMeter;

	private final Metric<Timer> localTaskSubmitTimer;
	private final Metric<Timer> taskExecutionTimer;

	private final Metric<Timer> getGroupSizesTimer;
	private final Metric<Timer> getOldestTaskTimeTimer;
	private final Metric<Timer> getQueueSizeTimer;

	private final Metric<Histogram> futureWaitTimeHistogram;

	private final Metric<Meter> taskErrors;
	private final Metric<Meter> routesSkipped;
	private final Metric<Meter> routeNotFound;
	private final Metric<Timer> taskQueuePollTimer;

	private final Metric<Counter> noRebalanceToDo;
	private final Metric<Meter> recoveryMeter;

	private final Metric<Timer> removeFromWriteAheadLogTimer;
	private final Metric<Timer> taskFinishedNotificationTimer;

	private final Metric<Timer> getReadyMemberTimer;

	private final Metric<Timer> findFailedFuturesTimer;
	private final Metric<Counter> failedFuturesCount;

	public ExecutorMetrics(HazeltaskConfig<?> config) {
		this.topologyName = config.getTopologyName();
		this.metrics = config.getMetricsRegistry();

		String name = createMetricName(TaskRecoveryTimerTask.class, "flush-timer");

		staleTaskFlushTimer = new Metric<Timer>(name, metrics.timer(name));

		// StaleWorkFlushTimerTask metrics
		name = createMetricName(TaskRecoveryTimerTask.class, "task-recovered");
		staleFlushCountHistogram = new Metric<Histogram>(name, metrics.histogram(name));

		// Work rebalancing metrics
		name = createMetricName(TaskRebalanceTimerTask.class,
				"tasks-redistributed");
		taskBalanceHistogram = new Metric<Histogram>(name, metrics.histogram(name));

		name = createMetricName(TaskRebalanceTimerTask.class, "balance-time");
		taskBalanceTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(TaskRebalanceTimerTask.class, "lock-wait-time");
		taskBalanceLockWaitTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(DistributedExecutorService.class,
				"task-submit-time");
		taskSubmitTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(DistributedExecutorService.class,
				"task-rejected-meter");
		taskRejectedMeter = new Metric<Meter>(name, metrics.meter(name));

		name = createMetricName(LocalTaskExecutorService.class,
				"task-submitted");
		localTaskSubmitTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(LocalTaskExecutorService.class, "task-executed");
		taskExecutionTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(LocalTaskExecutorService.class,
				"getGroupSizes-timer");
		getGroupSizesTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(LocalTaskExecutorService.class,
				"getOldestTaskTime-timer");
		getOldestTaskTimeTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(LocalTaskExecutorService.class,
				"getQueueSize-time");
		getQueueSizeTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(DistributedFutureTracker.class,
				"future-wait-time");
		futureWaitTimeHistogram = new Metric<Histogram>(name,
				metrics.histogram(name));

		name = createMetricName(LocalTaskExecutorService.class, "task-errors");
		taskErrors = new Metric<Meter>(name, metrics.meter(name));

		name = createMetricName("GroupedPriorityQueue", "routes-skipped");
		routesSkipped = new Metric<Meter>(name, metrics.meter(name));

		name = createMetricName("GroupedPriorityQueue", "route-not-found");
		routeNotFound = new Metric<Meter>(name, metrics.meter(name));

		name = createMetricName("GroupedPriorityQueue", "poll-time");
		taskQueuePollTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(TaskRebalanceTimerTask.class,
				"task-rebalance-noop");
		noRebalanceToDo = new Metric<Counter>(name, metrics.counter(name));

		name = createMetricName(TaskRecoveryTimerTask.class, "task-recovery");
		recoveryMeter = new Metric<Meter>(name, metrics.meter(name));

		name = createMetricName(LocalTaskExecutorService.class,
				"remove-from-write-ahead-log-time");
		removeFromWriteAheadLogTimer = new Metric<Timer>(name,
				metrics.timer(name));

		name = createMetricName(LocalTaskExecutorService.class,
				"task-finished-notification-time");
		taskFinishedNotificationTimer = new Metric<Timer>(name,
				metrics.timer(name));

		// FIXME: this metric doesn't belong here
		name = createMetricName(HazeltaskTopologyService.class,
				"getReadyMembers-time");
		getReadyMemberTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(DistributedFutureTracker.class,
				"find-failed-futures-time");
		findFailedFuturesTimer = new Metric<Timer>(name, metrics.timer(name));

		name = createMetricName(DistributedFutureTracker.class,
				"failed-futures-count");
		failedFuturesCount = new Metric<Counter>(name, metrics.counter(name));
	}

	public Metric<Timer> getRecoveryTimer() {
		return staleTaskFlushTimer;
	}

	public Metric<Histogram> getStaleFlushCountHistogram() {
		return staleFlushCountHistogram;
	}

	public Metric<Timer> getTaskBalanceTimer() {
		return taskBalanceTimer;
	}

	public Metric<Histogram> getTaskBalanceHistogram() {
		return taskBalanceHistogram;
	}

	public Metric<Timer> getTaskBalanceLockWaitTimer() {
		return taskBalanceLockWaitTimer;
	}

	public Metric<Timer> getTaskSubmitTimer() {
		return taskSubmitTimer;
	}

	public Metric<Meter> getTaskRejectedMeter() {
		return taskRejectedMeter;
	}

	private String createMetricName(Class<?> clz, String name) {
		return createMetricName(clz.getSimpleName(), name);
	}

	private String createMetricName(String className, String name) {
		return "hazeltask."+topologyName+"."+className+"."+name;
	}

	public Metric<Timer> getLocalTaskSubmitTimer() {
		return localTaskSubmitTimer;
	}

	public Metric<Timer> getTaskExecutionTimer() {
		return taskExecutionTimer;
	}

	public Metric<Timer> getGetGroupSizesTimer() {
		return getGroupSizesTimer;
	}

	public Metric<Timer> getGetOldestTaskTimeTimer() {
		return getOldestTaskTimeTimer;
	}

	public Metric<Timer> getGetQueueSizeTimer() {
		return getQueueSizeTimer;
	}

	public Metric<Histogram> getFutureWaitTimeHistogram() {
		return futureWaitTimeHistogram;
	}

	/**
	 * Calling this twice will not actually overwrite the gauge
	 * 
	 * @param collectionSizeGauge
	 */
	public void registerCollectionSizeGauge(
			CollectionSizeGauge collectionSizeGauge) {
		String name = createMetricName(LocalTaskExecutorService.class,
				"queue-size");
		metrics.register(name, collectionSizeGauge);
	}

	public void registerExecutionThroughputGauge(
			TaskThroughputGauge throughputGauge) {
		String name = createMetricName(LocalTaskExecutorService.class,
				"throughput");
		metrics.register(name, throughputGauge);
	}

	public void registerLocalFuturesWaitingGauge(LocalFuturesWaitingGauge gauge) {
		String name = createMetricName(DistributedExecutorService.class,
				"futures-waiting-count");
		metrics.register(name, gauge);
	}

	public void registerLocalWriteAheadLogSizeGauge(Gauge<Integer> gauge) {
		String name = createMetricName(DistributedExecutorService.class,
				"write-ahead-log-size");
		metrics.register(name, gauge);
	}

	public Metric<Meter> getTaskErrors() {
		return taskErrors;
	}

	public Metric<Meter> getRoutesSkipped() {
		return routesSkipped;
	}

	public Metric<Meter> getRouteNotFound() {
		return routeNotFound;
	}

	public Metric<Counter> getRebalanceNoopCounter() {
		return noRebalanceToDo;
	}

	public Metric<Meter> getRecoveryMeter() {
		return recoveryMeter;
	}

	public Metric<Timer> getRemoveFromWriteAheadLogTimer() {
		return removeFromWriteAheadLogTimer;
	}

	public Metric<Timer> getTaskFinishedNotificationTimer() {
		return taskFinishedNotificationTimer;
	}

	public Metric<Timer> getGetReadyMemberTimer() {
		return getReadyMemberTimer;
	}

	public Metric<Timer> getTaskQueuePollTimer() {
		return taskQueuePollTimer;
	}

	public Metric<Timer> getFindFailedFuturesTimer() {
		return findFailedFuturesTimer;
	}

	public Metric<Counter> getFailedFuturesCount() {
		return failedFuturesCount;
	}

} @Slf4j
public class TaskRebalanceTimerTask<GROUP extends Serializable> extends BackoffTask {
    private final Member localMember;
    private final IExecutorTopologyService<GROUP> executorTopologyService;
    private final LocalTaskExecutorService<GROUP> localSvc;
    
    private Histogram histogram;
    private Timer redistributionTimer;
    private Timer lockWaitTimer;
    private Counter getRebalanceNoopCounter;
	
	/**
	 * If a member has this percent MORE works than the current member, steal them
	 */
	private static final double THRESHOLD = 0.30;
	
	private final Lock LOCK;
	
	public TaskRebalanceTimerTask(HazeltaskTopology<GROUP> topology, LocalTaskExecutorService<GROUP> localSvc, IExecutorTopologyService<GROUP> executorTopologyService, ExecutorMetrics metrics) {
	    LOCK = executorTopologyService.getRebalanceTaskClusterLock();
		localMember = topology.getLocalMember();
		this.executorTopologyService = executorTopologyService;
		this.localSvc = localSvc;
		
		histogram = metrics.getTaskBalanceHistogram().getMetric();
        redistributionTimer = metrics.getTaskBalanceTimer().getMetric();
        lockWaitTimer = metrics.getTaskBalanceLockWaitTimer().getMetric();
        getRebalanceNoopCounter = metrics.getRebalanceNoopCounter().getMetric();
	}
	
	
	@Override
    public boolean execute() {
	    try {
    	    log.debug( "Running Rebalance Task");
    	    Timer.Context waitCtx = lockWaitTimer.time();
    	    try {    	        
    	        LOCK.lock();
    	    } finally {
    	        //TODO: refactor this so its cleaner
    	        try {
    	            waitCtx.stop();
    	        } catch (Throwable t) {
    	            //not sure if this is possible
    	            LOCK.unlock();
    	            throw t;//outer catch should get this.
    	        }
    	    }
    	    /*
    	     * NOTE: because we are locking here, we need to make ABSOLUTELY sure all our waits are bounded
    	     * ----> Comment on any external calls to note that they are bounded
    	     */
    	    Timer.Context timerCtx = null; 
    	    try {
    	        timerCtx = redistributionTimer.time();
    	        //ClusterServices clusterServices = distributedExecutorService.getTopology().getClusterServices();
        		
        	    //BOUNDED: MemberTasks.executeOptimistic waits a max of 60 seconds
        	    Collection<MemberResponse<Long>> queueSizes = executorTopologyService.getMemberQueueSizes();
        	    if(queueSizes.size() == 0) {
        	        getRebalanceNoopCounter.inc();
        	        return false;
        	    }
        		
                long localQueueSize = -1;        
        		long totalSize = 0;
        	
        		for(MemberResponse<Long> response : queueSizes) {
        			totalSize += response.getValue();
        			if(response.getMember().equals(localMember)) {
        			    localQueueSize = response.getValue();
        			}
        		}
        		
        		final long optimalSize = totalSize / queueSizes.size();
        		
        		if(localQueueSize == -1) {
        		    //the localQueueSize was not fetched for some reason... 
        		    //TODO: throw exception?
        		    log.error( "Cannot get localQueueSize");
        		    return false;
        		}	
        		
        		if(localQueueSize >= optimalSize || (optimalSize * THRESHOLD) <= localQueueSize) {
        		    //nothing to do
        		    log.info( "No rebalance needed");
        		    getRebalanceNoopCounter.inc();
        		    return false;
        		}
        		
        		//------------------------
        		// Figure out how much is available to take.. (sum of those that exceed the average size)
        		//------------------------
        		long totalExceedingIdeal = 0;
        		for(MemberResponse<Long> response : queueSizes) {
        		    if(response.getValue() > optimalSize)
        		        totalExceedingIdeal += response.getValue();
        		}
        		
        		final long needToTake =  optimalSize - localQueueSize;
        		LinkedList<MemberValuePair<Long>> numToTake = new LinkedList<MemberValuePair<Long>>();
        		
        		log.info( "Total Size: "+totalSize+", Optimal size: "+optimalSize+", Local Size: "+localQueueSize);
        		log.info( "I will take "+needToTake+" tasks from "+numToTake.size()+" nodes");
        		
        		//------------------------
        		// Take a percentage according to the total available to take
        		//------------------------
        		for(MemberResponse<Long> response : queueSizes) {
        		    if(response.getValue() > optimalSize) {
        		        double percent = ((double)response.getValue() / (double)totalExceedingIdeal);
        		        long take = (long) Math.round(needToTake * percent);
        		    	numToTake.add(new MemberValuePair<Long>(response.getMember(), take));
        		    	log.info( "I will take "+take+" tasks from "+response.getMember());
        		    }
        		}
    		
        		
    		//for each numToTake, send a message to steal work
    		//use a completion service to manage futures and recieve results
        	//make sure to bound the waiting of each call with something like 5 minutes or 10 minutes
        		
        		//TODO: replace this with a completion service so we can process results as we get them
        		Collection<HazeltaskTask<GROUP>> stolenTasks = executorTopologyService.stealTasks(numToTake);
        		//add to local queue
        		int totalAdded = 0;
        		for(HazeltaskTask<GROUP> task : stolenTasks) {
        		    localSvc.execute(task);
        		    totalAdded++;
        		}
        		
        		if(histogram != null)
        		    histogram.update(totalAdded);
        		
        		log.info( "Done adding "+totalAdded+"...");
        		
        		
    	    } finally {
    	        try {
    	            LOCK.unlock();
    	        } finally {
    	            timerCtx.stop();
    	        }
    	    }
            return false;
	    } catch (Throwable t) {
	        //catch all exceptions and swallow so it doens't cancel our timer task
	        log.error( "Error running Rebalance Task", t);
	        return true;
	    }
	}
}
 @Slf4j
public class TaskRecoveryTimerTask<GROUP extends Serializable> extends BackoffTask {
    private final DistributedExecutorServiceImpl<GROUP> svc;
    private final IExecutorTopologyService<GROUP> executorTopologyService;

    public static long EXPIRE_TIME_BUFFER = 5000L; //5 seconds
    public static long EMPTY_EXPIRE_TIME_BUFFER = 10000L; //10 seconds
    
    private Timer flushTimer;
    private Histogram numFlushedHistogram;
    private Meter recoveryMeter;
    
    public TaskRecoveryTimerTask(HazeltaskTopology<GROUP> topology, DistributedExecutorServiceImpl<GROUP> svc, IExecutorTopologyService<GROUP> executorTopologyService, ExecutorMetrics metrics) {
        this.svc = svc;
        this.flushTimer = metrics.getRecoveryTimer().getMetric();
        this.numFlushedHistogram = metrics.getStaleFlushCountHistogram().getMetric();
        this.executorTopologyService = executorTopologyService;
        this.recoveryMeter = metrics.getRecoveryMeter().getMetric();
    }

    @Override
    public boolean execute() {
    	try {
            boolean flushed = false;
            Timer.Context timerCtx = flushTimer.time();
        	
        	try {
  
        	    Collection<MemberResponse<Long>> results = executorTopologyService.getOldestTaskTimestamps();
    	
    	        long min = Long.MAX_VALUE;
    	        for(MemberResponse<Long> result : results) {
    	            if(result.getValue() != null && result.getValue() < min) {
    	                min = result.getValue();
    	            }
    	        }
    	        
    	        String sql;
    	        if(min == Long.MAX_VALUE) {
    	            sql = "createdAtMillis < "+(System.currentTimeMillis()-EMPTY_EXPIRE_TIME_BUFFER);
    	        } else {
    	            sql = "createdAtMillis < "+(min-EXPIRE_TIME_BUFFER);
    	        }
    	        
    	        //System.out.println("Map Size: "+map.size()+" "+map.values().size());
    	        //System.out.println("Local Size: "+map.localKeySet().size());
    	        
    //	        Set<String> keys = (Set<String>) map.localKeySet(pred);
    //	        Collection<HazelcastWork> works = map.getAll(keys).values();
    	        
    	        Collection<HazeltaskTask<GROUP>> works = executorTopologyService.getLocalPendingTasks(sql);
    	        
    	        if(works.size() > 0) {
    	            flushed = true;
    	            recoveryMeter.mark();
    	            log.info("Recovering "+works.size()+" works. "+sql);
    	        }
    	        
    	        for(HazeltaskTask<GROUP> work : works) {
    	            svc.submitHazeltaskTask(work, true);
    	        }
    	        
    	        if(works.size() > 0)
    	        	log.info("Done recovering "+works.size()+" works");
    	        
    	        numFlushedHistogram.update(works.size());
    	        
        	} finally {
        		timerCtx.stop();
        	}
        	
        	return flushed;
    	} catch(Throwable t) {
    	    //swallow this exception so it doens't cancel this task
    	    log.error("An error occurred while trying to recover tasks", t);
    	    return true;//backoff
    	}
    }
} @Slf4j
public class HazelcastPartitionManager {
    private final CopyOnWriteArrayList<PartitionLostListener> listeners = new CopyOnWriteArrayList<PartitionLostListener>();
    private final PartitionService partitionService; 
    
    
    public HazelcastPartitionManager(PartitionService partitionService) {
        this.partitionService = partitionService;
        partitionService.addMigrationListener(new MigrationListener() {
            
            @Override
            public void migrationStarted(MigrationEvent migrationEvent) {}
            
            @Override
            public void migrationFailed(MigrationEvent migrationEvent) {
                migrationCompleted(migrationEvent);
            }
            
            @Override
            public void migrationCompleted(MigrationEvent migrationEvent) {
                if(migrationEvent.getOldOwner() == null) {
                    for(PartitionLostListener listener : listeners) {
                        try {
                            listener.partitionLost(migrationEvent);
                        } catch (Exception e) {
                            //swallow
                            log.error("An exception was thrown by our partitionLost event listener.  I will ignore it. ", e);
                        }
                    }
                }
            }
        });
    }
    
    public Partition getPartition(UUID id) {
    	return partitionService.getPartition(id);
    }
    
    public void addPartitionListener(PartitionLostListener listener) {
        listeners.add(listener);
    }
    
    public static interface PartitionLostListener {
        public abstract void partitionLost(MigrationEvent migrationEvent);
    }
} public static interface PartitionLostListener {
    public abstract void partitionLost(MigrationEvent migrationEvent);
}
import java.io.Serializable;
import java.util.Set;
import java.util.UUID;
import lombok.extern.slf4j.Slf4j;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.LifecycleEvent;
import com.hazelcast.core.LifecycleEvent.LifecycleState;
import com.hazelcast.core.LifecycleListener;
import com.hazelcast.core.LifecycleService;
import com.hazelcast.core.MemberLeftException;
import com.hazelcast.core.MigrationEvent;
import com.hazelcast.core.Partition;
import com.hazelcast.core.PartitionService;
import com.hazeltask.config.ConfigValidator;
import com.hazeltask.config.ExecutorConfig;
import com.hazeltask.config.HazeltaskConfig;
import com.hazeltask.core.concurrent.BackoffTimer;
import com.hazeltask.executor.DistributedExecutorService;
import com.hazeltask.executor.DistributedExecutorServiceImpl;
import com.hazeltask.executor.DistributedFutureTracker;
import com.hazeltask.executor.HazelcastExecutorTopologyService;
import com.hazeltask.executor.IExecutorTopologyService;
import com.hazeltask.executor.TaskResponseListener;
import com.hazeltask.executor.local.LocalTaskExecutorService;
import com.hazeltask.executor.metrics.ExecutorMetrics;
import com.hazeltask.executor.task.TaskRebalanceTimerTask;
import com.hazeltask.executor.task.TaskRecoveryTimerTask;
import com.hazeltask.hazelcast.HazelcastPartitionManager;
import com.hazeltask.hazelcast.HazelcastPartitionManager.PartitionLostListener;
package com.hazeltask;




/**
 * TODO: this class is messy... clean it up
 * @author jclawson
 *
 * @param <GROUP>
 */
@Slf4j
public class HazeltaskInstance<GROUP extends Serializable> {
    
    private final DistributedExecutorServiceImpl<GROUP>       executor;
    private final HazeltaskTopology<GROUP>                topology;
    private final HazeltaskConfig<GROUP> hazeltaskConfig;
    private final UUID hazeltaskInstanceId = UUID.randomUUID();
    private final ExecutorConfig<GROUP> executorConfig;
    
    private final ClusterService<GROUP> clusterService;
    private final ITopologyService<GROUP> topologyService;
    private final IExecutorTopologyService<GROUP> executorTopologyService;
    private final LocalTaskExecutorService<GROUP> localExeutorService;
    private final ExecutorMetrics executorMetrics;
    
    /**
     * @param hazeltaskConfig
     */
    protected HazeltaskInstance(HazeltaskConfig<GROUP> hazeltaskConfig) {
        this.hazeltaskConfig = hazeltaskConfig;
        
        ConfigValidator.validate(hazeltaskConfig);
        
        executorConfig = hazeltaskConfig.getExecutorConfig();
        HazelcastInstance hazelcast = hazeltaskConfig.getHazelcast();
        final String topologyName = hazeltaskConfig.getTopologyName();
        executorMetrics = new ExecutorMetrics(hazeltaskConfig);
        topologyService = new HazeltaskTopologyService<GROUP>(hazeltaskConfig, executorMetrics.getGetReadyMemberTimer().getMetric());
        
        
        PartitionService partitionService = hazelcast.getPartitionService();
        
        this.topology = new HazeltaskTopology<GROUP>(topologyName, hazelcast.getCluster().getLocalMember());
        executorTopologyService = new HazelcastExecutorTopologyService<GROUP>(hazeltaskConfig, topology);
        clusterService = new HazeltaskStatisticsService<GROUP>(executorTopologyService);
        
        if(!executorConfig.isDisableWorkers())
            localExeutorService = new LocalTaskExecutorService<GROUP>(hazelcast, executorConfig, hazeltaskConfig.getThreadFactory(), executorTopologyService, executorMetrics);
        else
            localExeutorService = null;
        
        final DistributedFutureTracker<GROUP> futureTracker;
        

        if(executorConfig.isFutureSupportEnabled()) {
                       
            futureTracker = new DistributedFutureTracker<GROUP>(executorTopologyService, executorMetrics, executorConfig);
            final HazelcastPartitionManager partitionManager = new HazelcastPartitionManager(partitionService);
            
            /*
             * TODO: we can make losing partitions nicer:
             * 1) don't loop so much... this is like o(p*n).. we can make it o(n)
             * 2) optionally have the future listener store the HazeltaskTask its watching.  It can re-add
             *    the task if the partition is lost.  (possible race condition here-- would double do work)
             * 3) organize futures in 2 hashmaps... partitionId -> uuid -> future.
             */
            partitionManager.addPartitionListener(new PartitionLostListener() {
                @Override
                public void partitionLost(MigrationEvent migrationEvent) {
                    //TODO: make this faster, too many loops
                    Set<UUID> uuids = futureTracker.getTrackedTaskIds();
                    for(UUID uuid : uuids) {
                        Partition partition = partitionManager.getPartition(uuid);
                        if(migrationEvent.getPartitionId() == partition.getPartitionId()) {
                            futureTracker.errorFuture(uuid, new MemberLeftException());
                        }
                    }
                }
            });
            
            executorTopologyService.addTaskResponseMessageHandler(futureTracker);
            for(TaskResponseListener listener : executorConfig.getTaskResponseListeners()) {
                executorTopologyService.addTaskResponseMessageHandler(listener);
            }
            
        } else {
            futureTracker = null;
        }
        
        executor = new DistributedExecutorServiceImpl<GROUP>(topology, executorTopologyService, executorConfig, futureTracker, localExeutorService, executorMetrics);
        
    }
    
    protected void start() {
        BackoffTimer hazeltaskTimer = new BackoffTimer(hazeltaskConfig.getTopologyName(), hazeltaskConfig.getThreadFactory().named("timertasks"));
        setupDistributedExecutor(hazeltaskConfig.getHazelcast(), topology, hazeltaskTimer, executorConfig, executor, topologyService, executorTopologyService, localExeutorService, executorMetrics);
        
        //if autoStart... we need to start
        if(executorConfig.isAutoStart()) {
            final LifecycleService lifecycleService = hazeltaskConfig.getHazelcast().getLifecycleService();
            LifecycleListener autoStartListener = new LifecycleListener() {                
                public void stateChanged(LifecycleEvent event) {
                    if(event.getState() == LifecycleState.STARTED) {
                        log.info(topology.getName()+" Hazeltask instance is starting up due to Hazelcast startup");
                        executor.startup();
                    } else if (event.getState() == LifecycleState.SHUTTING_DOWN) {
                        log.info(topology.getName()+" Hazeltask instance is shutting down due to Hazelcast shutdown");
                        executor.shutdown();
                    }
                }
            };
            lifecycleService.addLifecycleListener(autoStartListener);
            
            if(lifecycleService.isRunning()) {
                log.info(topology.getName()+" Hazeltask instance is starting up");
                executor.startup();
            }
        }
    }
    
    private void setupDistributedExecutor(final HazelcastInstance hazelcast, final HazeltaskTopology<GROUP> topology, final BackoffTimer hazeltaskTimer, final ExecutorConfig<GROUP> executorConfig, DistributedExecutorServiceImpl<GROUP> svc, ITopologyService<GROUP> topologySvc, IExecutorTopologyService<GROUP> executorTopologyService, LocalTaskExecutorService<GROUP> localExeutorService, ExecutorMetrics executorMetrics) {
        final TaskRecoveryTimerTask<GROUP> bundleTask = new TaskRecoveryTimerTask<GROUP>(topology, svc, executorTopologyService, executorMetrics);
        final TaskRebalanceTimerTask<GROUP> rebalanceTask;
        if(!svc.getExecutorConfig().isDisableWorkers())
            rebalanceTask = new TaskRebalanceTimerTask<GROUP>(topology, localExeutorService, executorTopologyService, executorMetrics);
        else
            rebalanceTask = null;
        final IsMemberReadyTimerTask<GROUP> getReadyMembersTask = new IsMemberReadyTimerTask<GROUP>(topologySvc, topology);
        
        //execute the getReadyMembers task immediately
        hazeltaskTimer.schedule(getReadyMembersTask, 20000, 20000);
        getReadyMembersTask.execute();
   
        hazeltaskConfig.getHazelcast().getCluster().addMembershipListener(getReadyMembersTask);
        
        svc.addServiceListener(new HazeltaskServiceListener<DistributedExecutorService<GROUP>>(){