Refactor BalancedMetricResolver to allow HDFS and FILE backing, add callbacks for major events, begin working on tests

Refactor BalancedMetricResolver, allow FILE or HDFS persister, add callbacks for events, begin working on testing

Fixes

balancer/pom.xml

@@ -108,6 +108,10 @@
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-minicluster</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.glassfish.tyrus</groupId>
+      <artifactId>tyrus-client</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>

balancer/src/main/java/timely/balancer/ArrivalRate.java

@@ -1,7 +1,7 @@
 package timely.balancer;
 
-import java.util.Timer;
+import java.util.concurrent.ScheduledExecutorService;
-import java.util.TimerTask;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
 import com.google.common.util.concurrent.AtomicDouble;
@@ -20,38 +20,26 @@ public class ArrivalRate {
     private AtomicDouble longTermRate = new AtomicDouble(0);
     private AtomicLong longTermLastReset = new AtomicLong(0);
 
-    public ArrivalRate(Timer timer) {
+    public ArrivalRate(ScheduledExecutorService arrivalRateExecutor) {
         long now = System.currentTimeMillis();
         shortTermLastReset.set(now);
         mediumTermLastReset.set(now);
         longTermLastReset.set(now);
 
         // 2 minute
-        timer.schedule(new TimerTask() {
+        arrivalRateExecutor.scheduleAtFixedRate(() -> {
-
+            resetShort();
-            @Override
+        }, 2, 2, TimeUnit.MINUTES);
-            public void run() {
-                resetShort();
-            }
-        }, 120000, 120000);
 
         // 5 minute
-        timer.schedule(new TimerTask() {
+        arrivalRateExecutor.scheduleAtFixedRate(() -> {
-
+            resetMedium();
-            @Override
+        }, 5, 5, TimeUnit.MINUTES);
-            public void run() {
-                resetMedium();
-            }
-        }, 300000, 300000);
 
         // 15 minute
-        timer.schedule(new TimerTask() {
+        arrivalRateExecutor.scheduleAtFixedRate(() -> {
-
+            resetLong();
-            @Override
+        }, 15, 15, TimeUnit.MINUTES);
-            public void run() {
-                resetLong();
-            }
-        }, 900000, 900000);
     }
 
     private void resetShort() {

balancer/src/main/java/timely/balancer/Balancer.java

@@ -77,7 +77,7 @@ import timely.balancer.connection.http.HttpClientPool;
 import timely.balancer.connection.tcp.TcpClientPool;
 import timely.balancer.connection.udp.UdpClientPool;
 import timely.balancer.connection.ws.WsClientPool;
-import timely.balancer.healthcheck.HealthChecker;
+import timely.balancer.healthcheck.TcpHealthChecker;
 import timely.balancer.netty.http.HttpRelayHandler;
 import timely.balancer.netty.tcp.TcpRelayHandler;
 import timely.balancer.netty.udp.UdpRelayHandler;
@@ -145,8 +145,11 @@ public class Balancer {
         this.balancerConfig = balancerConf;
     }
 
-    protected static ConfigurableApplicationContext initializeConfiguration(String[] args) {
+    protected static ConfigurableApplicationContext initializeContext(String[] args) {
-        return new SpringApplicationBuilder(SpringBootstrap.class).web(WebApplicationType.NONE).run(args);
+        SpringApplicationBuilder builder = new SpringApplicationBuilder(SpringBootstrap.class);
+        builder.web(WebApplicationType.NONE);
+        builder.registerShutdownHook(false);
+        return builder.run(args);
     }
 
     private void shutdownHook() {
@@ -178,8 +181,9 @@ public class Balancer {
             channelFutures.add(wsChannelHandle.close());
         }
 
+        int udpChannel = 1;
         for (Channel c : udpChannelHandleList) {
-            LOG.info("Closing udpChannelHandle");
+            LOG.info("Closing udpChannelHandle #" + udpChannel++);
             channelFutures.add(c.close());
         }
 
@@ -188,14 +192,13 @@ public class Balancer {
             try {
                 f.get();
             } catch (final Exception e) {
-                LOG.error("Error while shutting down channel: {}", f.channel().config());
+                LOG.error("Channel:" + f.channel().config() + " -> " + e.getMessage(), e);
-                LOG.error("{}", e.getMessage());
-                e.printStackTrace();
             }
         });
 
-        Integer quietPeriod = balancerConfig.getServer().getShutdownQuietPeriod();
+        int quietPeriod = balancerConfig.getServer().getShutdownQuietPeriod();
         List<Future<?>> groupFutures = new ArrayList<>();
+
         if (tcpBossGroup != null) {
             LOG.info("Shutting down tcpBossGroup");
             groupFutures.add(tcpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
@@ -240,9 +243,7 @@ public class Balancer {
             try {
                 f.get();
             } catch (final Exception e) {
-                LOG.error("Error while shutting down group: {}", f.toString());
+                LOG.error("Group:" + f.toString() + " -> " + e.getMessage(), e);
-                LOG.error("{}", e.getMessage());
-                e.printStackTrace();
             }
         });
 
@@ -253,14 +254,29 @@ public class Balancer {
             LOG.error("Error closing WebSocketRequestDecoder during shutdown", e);
         }
 
+        if (this.metricResolver != null) {
+            try {
+                LOG.info("Closing metricResolver");
+                this.metricResolver.stop();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+
+        if (curatorFramework != null) {
+            try {
+                LOG.info("Closing curatorFramework");
+                curatorFramework.close();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+
         if (applicationContext != null) {
             LOG.info("Closing applicationContext");
             applicationContext.close();
         }
 
-        if (curatorFramework != null) {
-            curatorFramework.close();
-        }
         this.shutdown = true;
         LOG.info("Server shut down.");
     }
@@ -401,8 +417,9 @@ public class Balancer {
         for (int x = 0; x < balancerConfig.getServer().getNumTcpPools(); x++) {
             tcpClientPools.add(new TcpClientPool(this.balancerConfig));
         }
-        HealthChecker healthChecker = new HealthChecker(this.balancerConfig, tcpClientPools.get(0));
+        TcpHealthChecker healthChecker = new TcpHealthChecker(this.balancerConfig, tcpClientPools.get(0));
         this.metricResolver = new BalancedMetricResolver(curatorFramework, this.balancerConfig, healthChecker);
+        this.metricResolver.start();
 
         final ServerBootstrap tcpServer = new ServerBootstrap();
         tcpServer.group(tcpBossGroup, tcpWorkerGroup);
@@ -481,12 +498,12 @@ public class Balancer {
 
     public static void main(String[] args) throws Exception {
 
-        Balancer.applicationContext = Balancer.initializeConfiguration(args);
+        Balancer.applicationContext = Balancer.initializeContext(args);
-        BalancerConfiguration balancerConf = applicationContext.getBean(BalancerConfiguration.class);
+        BalancerConfiguration balancerConf = Balancer.applicationContext.getBean(BalancerConfiguration.class);
 
-        Balancer b = new Balancer(balancerConf);
+        Balancer balancer = new Balancer(balancerConf);
         try {
-            b.run();
+            balancer.run();
             LATCH.await();
         } catch (final InterruptedException e) {
             LOG.info("Server shutting down.");
@@ -494,7 +511,7 @@ public class Balancer {
             LOG.error("Error running server.", e);
         } finally {
             try {
-                b.shutdown();
+                balancer.shutdown();
             } catch (Exception e) {
                 System.exit(1);
             }
@@ -518,11 +535,13 @@ public class Balancer {
                 ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
                 ch.pipeline().addLast("queryDecoder", new timely.netty.http.HttpRequestDecoder(
                         balancerConfig.getSecurity(), balancerConfig.getHttp()));
-                ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler());
+                ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler().setIgnoreSslHandshakeErrors(
+                        balancerConfig.getSecurity().getServerSsl().isUseGeneratedKeypair()));
                 ch.pipeline().addLast("login",
                         new X509LoginRequestHandler(balancerConfig.getSecurity(), balancerConfig.getHttp()));
                 ch.pipeline().addLast("httpRelay", new HttpRelayHandler(metricResolver, httpClientPool));
-                ch.pipeline().addLast("error", new TimelyExceptionHandler());
+                ch.pipeline().addLast("error", new TimelyExceptionHandler().setIgnoreSslHandshakeErrors(
+                        balancerConfig.getSecurity().getServerSsl().isUseGeneratedKeypair()));
             }
         };
     }

balancer/src/main/java/timely/balancer/configuration/BalancerConfiguration.java

@@ -49,6 +49,16 @@ public class BalancerConfiguration {
 
     private int serverSuccessesBeforeUp = 3;
 
+    private String metricAssignmentPersisterType = "HDFS";
+
+    private long balanceDelay = 900000;
+
+    private long balancePeriod = 900000;
+
+    private long persistDelay = 10000;
+
+    private long persistPeriod = 60000;
+
     public ZooKeeper getZooKeeper() {
         return zooKeeper;
     }
@@ -120,4 +130,44 @@ public class BalancerConfiguration {
     public void setServerSuccessesBeforeUp(int serverSuccessesBeforeUp) {
         this.serverSuccessesBeforeUp = serverSuccessesBeforeUp;
     }
+
+    public String getMetricAssignmentPersisterType() {
+        return metricAssignmentPersisterType;
+    }
+
+    public void setMetricAssignmentPersisterType(String metricAssignmentPersisterType) {
+        this.metricAssignmentPersisterType = metricAssignmentPersisterType;
+    }
+
+    public void setBalanceDelay(long balanceDelay) {
+        this.balanceDelay = balanceDelay;
+    }
+
+    public long getBalanceDelay() {
+        return balanceDelay;
+    }
+
+    public void setBalancePeriod(long balancePeriod) {
+        this.balancePeriod = balancePeriod;
+    }
+
+    public long getBalancePeriod() {
+        return balancePeriod;
+    }
+
+    public void setPersistDelay(long persistDelay) {
+        this.persistDelay = persistDelay;
+    }
+
+    public long getPersistDelay() {
+        return persistDelay;
+    }
+
+    public void setPersistPeriod(long persistPeriod) {
+        this.persistPeriod = persistPeriod;
+    }
+
+    public long getPersistPeriod() {
+        return persistPeriod;
+    }
 }

balancer/src/main/java/timely/balancer/connection/TimelyBalancedHost.java

@@ -1,7 +1,5 @@
 package timely.balancer.connection;
 
-import java.util.Timer;
-
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
@@ -13,7 +11,6 @@ import timely.balancer.configuration.BalancerConfiguration;
 public class TimelyBalancedHost {
 
     private static final Logger LOG = LoggerFactory.getLogger(TimelyBalancedHost.class);
-    static private Timer arrivalRateTimer = new Timer("ArrivalRateTimerTBH", true);
 
     private String host;
     private int tcpPort;
@@ -24,34 +21,29 @@ public class TimelyBalancedHost {
     private BalancerConfiguration balancerConfig;
     private int failures = 0;
     private int successes = 0;
-    private ArrivalRate arrivalRate = new ArrivalRate(arrivalRateTimer);
+    private ArrivalRate arrivalRate = null;
 
-    public TimelyBalancedHost() {
+    private TimelyBalancedHost() {
 
     }
 
-    public TimelyBalancedHost(String host, int tcpPort, int httpPort, int wsPort, int udpPort) {
+    private TimelyBalancedHost(String host, int tcpPort, int httpPort, int wsPort, int udpPort,
+            ArrivalRate arrivalRate) {
         this.host = host;
         this.tcpPort = tcpPort;
         this.httpPort = httpPort;
         this.wsPort = wsPort;
         this.udpPort = udpPort;
+        this.arrivalRate = arrivalRate;
     }
 
-    public TimelyBalancedHost(String host, int basePort) {
+    static public TimelyBalancedHost of(String host, int tcpPort, int httpPort, int wsPort, int udpPort,
-        this.host = host;
+            ArrivalRate arrivalRate) {
-        this.tcpPort = basePort;
+        return new TimelyBalancedHost(host, tcpPort, httpPort, wsPort, udpPort, arrivalRate);
-        this.httpPort = basePort + 1;
-        this.wsPort = basePort + 2;
-        this.udpPort = basePort + 3;
-    }
-
-    static public TimelyBalancedHost of(String host, int tcpPort, int httpPort, int wsPort, int udpPort) {
-        return new TimelyBalancedHost(host, tcpPort, httpPort, wsPort, udpPort);
     }
 
     static public TimelyBalancedHost of(String host, int basePort) {
-        return new TimelyBalancedHost(host, basePort);
+        return new TimelyBalancedHost(host, basePort, basePort + 1, basePort + 2, basePort + 3, null);
     }
 
     public void setBalancerConfig(BalancerConfiguration balancerConfig) {
@@ -141,11 +133,17 @@ public class TimelyBalancedHost {
     }
 
     public void arrived() {
-        arrivalRate.arrived();
+        if (this.arrivalRate != null) {
+            this.arrivalRate.arrived();
+        }
     }
 
     public double getArrivalRate() {
-        return arrivalRate.getRate();
+        if (this.arrivalRate == null) {
+            return 0;
+        } else {
+            return this.arrivalRate.getRate();
+        }
     }
 
     @Override
@@ -186,4 +184,11 @@ public class TimelyBalancedHost {
         builder.append(isUp);
         return builder.toString();
     }
+
+    public static String toStringShort(TimelyBalancedHost tbh) {
+        if (tbh == null) {
+            return null;
+        }
+        return tbh.host + ":" + tbh.tcpPort;
+    }
 }

balancer/src/main/java/timely/balancer/healthcheck/HealthChecker.java

@@ -1,53 +1,12 @@
 package timely.balancer.healthcheck;
 
-import java.util.ArrayList;
 import java.util.List;
-import java.util.Timer;
-import java.util.TimerTask;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.balancer.connection.tcp.TcpClientPool;
 
-public class HealthChecker {
+public interface HealthChecker {
 
-    private static final Logger LOG = LoggerFactory.getLogger(HealthChecker.class);
+    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts);
-    private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
-    private Timer timer;
-    final private TCPServerHealthCheck tcpServerHealthCheck;
-    private TimerTask check = new TimerTask() {
 
-        @Override
+    public void close() throws Exception;
-        public void run() {
-            synchronized (timelyHosts) {
-                for (TimelyBalancedHost h : timelyHosts) {
-                    try {
-                        if (tcpServerHealthCheck.isServerHealthy(h)) {
-                            h.reportSuccess();
-                        } else {
-                            h.reportFailure();
-                        }
-                    } catch (Exception e) {
-                        LOG.error(e.getMessage(), e);
-                        h.reportFailure();
-                    }
-                }
-            }
-        }
-    };
-
-    public HealthChecker(BalancerConfiguration config, TcpClientPool tcpClientPool) {
-        this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
-        this.timer = new Timer("HealthCheckerThread", true);
-        this.timer.scheduleAtFixedRate(check, 0, config.getCheckServerHealthInterval());
-    }
-
-    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
-        synchronized (timelyHosts) {
-            this.timelyHosts.clear();
-            this.timelyHosts.addAll(timelyHosts);
-        }
-    }
 }

balancer/src/main/java/timely/balancer/healthcheck/TcpHealthChecker.java

@@ -0,0 +1,62 @@
+package timely.balancer.healthcheck;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.balancer.configuration.BalancerConfiguration;
+import timely.balancer.connection.TimelyBalancedHost;
+import timely.balancer.connection.tcp.TcpClientPool;
+
+public class TcpHealthChecker implements HealthChecker {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TcpHealthChecker.class);
+    private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
+    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
+    final private TCPServerHealthCheck tcpServerHealthCheck;
+
+    public TcpHealthChecker(BalancerConfiguration config, TcpClientPool tcpClientPool) {
+        this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
+        this.executorService.scheduleAtFixedRate(() -> {
+            synchronized (timelyHosts) {
+                for (TimelyBalancedHost h : timelyHosts) {
+                    try {
+                        if (tcpServerHealthCheck.isServerHealthy(h)) {
+                            h.reportSuccess();
+                        } else {
+                            h.reportFailure();
+                        }
+                    } catch (Exception e) {
+                        LOG.error(e.getMessage(), e);
+                        h.reportFailure();
+                    }
+                }
+            }
+        }, 0, config.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
+        synchronized (timelyHosts) {
+            this.timelyHosts.clear();
+            this.timelyHosts.addAll(timelyHosts);
+        }
+    }
+
+    public void close() throws Exception {
+        this.executorService.shutdown();
+        try {
+            this.executorService.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.executorService.isTerminated()) {
+                this.executorService.shutdownNow();
+            }
+        }
+    }
+}

balancer/src/main/java/timely/balancer/resolver/BalancedMetricResolver.java

@@ -4,10 +4,14 @@ import static timely.Server.SERVICE_DISCOVERY_PATH;
 import static timely.balancer.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
 import static timely.balancer.Balancer.ASSIGNMENTS_LOCK_PATH;
 import static timely.balancer.Balancer.LEADER_LATCH_PATH;
+import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.ASSIGN_FALLBACK_SEQUENTIAL;
+import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.ASSIGN_ROUND_ROBIN;
+import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.HOST_DOWN_ROUND_ROBIN;
+import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.HOST_REMOVED;
+import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.REBALANCE;
 import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
 import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
 
-import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -19,21 +23,18 @@ import java.util.List;
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
-import com.csvreader.CsvReader;
-import com.csvreader.CsvWriter;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang3.SerializationUtils;
 import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
 import org.apache.curator.framework.recipes.atomic.DistributedAtomicValue;
 import org.apache.curator.framework.recipes.cache.TreeCache;
 import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
@@ -48,11 +49,6 @@ import org.apache.curator.x.discovery.ServiceDiscovery;
 import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
 import org.apache.curator.x.discovery.ServiceInstance;
 import org.apache.curator.x.discovery.details.ServiceCacheListener;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
 import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -61,55 +57,70 @@ import timely.balancer.ArrivalRate;
 import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.healthcheck.HealthChecker;
+import timely.balancer.resolver.eventing.MetricAssignedCallback;
+import timely.balancer.resolver.eventing.MetricAssignedEvent;
+import timely.balancer.resolver.eventing.MetricBalanceCallback;
+import timely.balancer.resolver.eventing.MetricBalanceEvent;
+import timely.balancer.resolver.eventing.MetricHostCallback;
+import timely.balancer.resolver.eventing.MetricHostEvent;
 
 public class BalancedMetricResolver implements MetricResolver {
 
     private static final Logger LOG = LoggerFactory.getLogger(BalancedMetricResolver.class);
     private Map<String, TimelyBalancedHost> metricToHostMap = new TreeMap<>();
+    private Set<String> unassignedMetrics = new HashSet<>();
     private Map<String, ArrivalRate> metricMap = new HashMap<>();
     private ReentrantReadWriteLock balancerLock = new ReentrantReadWriteLock();
     private List<TimelyBalancedHost> serverList = new ArrayList<>();
     private Random r = new Random();
-    final private HealthChecker healthChecker;
+    private final HealthChecker healthChecker;
-    private Timer timer = new Timer("RebalanceTimer", true);
+    private ScheduledExecutorService assignmentExecutor = Executors.newScheduledThreadPool(2);
-    private Timer arrivalRateTimer = new Timer("ArrivalRateTimerResolver", true);
+    private ScheduledExecutorService arrivalRateExecutor = Executors.newScheduledThreadPool(2);
     private int roundRobinCounter = 0;
     private Set<String> nonCachedMetrics = new HashSet<>();
     private DistributedAtomicValue nonCachedMetricsIP;
     private ReentrantReadWriteLock nonCachedMetricsLocalLock = new ReentrantReadWriteLock();
     private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
-    private BalancerConfiguration balancerConfig;
+    private AtomicLong assignmentsLastUpdatedLocal = new AtomicLong(0);
     private LeaderLatch leaderLatch;
     private AtomicBoolean isLeader = new AtomicBoolean(false);
     private InterProcessReadWriteLock assignmentsIPRWLock;
-    private DistributedAtomicLong assignmentsLastUpdatedInHdfs;
+    private MetricAssignmentPerister assignmentPerister;
-    private AtomicLong assignmentsLastUpdatedLocal = new AtomicLong(0);
 
-    private enum BalanceType {
+    final private BalancerConfiguration balancerConfig;
-        HIGH_LOW, HIGH_AVG, AVG_LOW
+    final private CuratorFramework curatorFramework;
+
+    private List<MetricAssignedCallback> metricAssignedCallbacks = new ArrayList<>();
+    private List<MetricBalanceCallback> metricBalanceCallbacks = new ArrayList<>();
+    private List<MetricHostCallback> metricHostCallbacks = new ArrayList<>();
+
+    public enum BalanceType {
+        HIGH_LOW, HIGH_AVG, AVG_LOW;
     }
 
     public BalancedMetricResolver(CuratorFramework curatorFramework, BalancerConfiguration balancerConfig,
             HealthChecker healthChecker) {
+        this.curatorFramework = curatorFramework;
         this.balancerConfig = balancerConfig;
         this.healthChecker = healthChecker;
+    }
 
+    public void start() {
         assignmentsIPRWLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
         testIPRWLock(curatorFramework, assignmentsIPRWLock, ASSIGNMENTS_LOCK_PATH);
+
+        this.assignmentPerister = new MetricAssignmentPerister(this, balancerConfig, metricToHostMap, nonCachedMetrics,
+                curatorFramework, assignmentsIPRWLock, nonCachedMetricsLocalLock, assignmentsLastUpdatedLocal,
+                balancerLock);
+
         startLeaderLatch(curatorFramework);
         startServiceListener(curatorFramework);
-        assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH,
-                new RetryForever(1000));
 
-        TreeCacheListener assignmentListener = new TreeCacheListener() {
+        TreeCacheListener assignmentListener = (curatorFramework1, event) -> {
-
+            LOG.debug("Handling assignments event {}. assignmentsLastUpdated:{}", event.getType().toString(),
-            @Override
+                    new Date(assignmentPerister.getAssignmentsLastUpdatedByLeader()));
-            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
+            if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
-                LOG.debug("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
+                assignmentPerister.readAssignments(true);
-                        new Date(assignmentsLastUpdatedInHdfs.get().postValue()));
-                if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
-                    readAssignmentsFromHdfs(true);
-                }
             }
         };
 
@@ -149,47 +160,38 @@ public class BalancedMetricResolver implements MetricResolver {
         }
         readNonCachedMetricsIP();
 
-        readAssignmentsFromHdfs(false);
+        assignmentPerister.readAssignments(false);
 
-        timer.schedule(new TimerTask() {
+        assignmentExecutor.scheduleAtFixedRate(() -> {
-
+            if (isLeader.get()) {
-            @Override
-            public void run() {
-                if (isLeader.get()) {
-                    try {
-                        balance();
-                    } catch (Exception e) {
-                        LOG.error(e.getMessage(), e);
-                    }
-                }
-            }
-        }, 900000, 900000);
-
-        timer.schedule(new TimerTask() {
-
-            @Override
-            public void run() {
                 try {
-                    if (isLeader.get()) {
+                    balance();
-                        long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
-                        long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
-                        if (lastLocalUpdate > lastHdfsUpdate) {
-                            LOG.debug("Leader writing assignments to hdfs lastLocalUpdate ({}) > lastHdfsUpdate ({})",
-                                    new Date(lastLocalUpdate), new Date(lastHdfsUpdate));
-                            writeAssignmentsToHdfs();
-                        } else {
-                            LOG.trace(
-                                    "Leader not writing assignments to hdfs lastLocalUpdate ({}) <= lastHdfsUpdate ({})",
-                                    new Date(lastLocalUpdate), new Date(lastHdfsUpdate));
-                        }
-                    } else {
-                        readAssignmentsFromHdfs(true);
-                    }
                 } catch (Exception e) {
                     LOG.error(e.getMessage(), e);
                 }
             }
-        }, 10000, 60000);
+        }, balancerConfig.getBalanceDelay(), balancerConfig.getBalancePeriod(), TimeUnit.MILLISECONDS);
+
+        assignmentExecutor.scheduleAtFixedRate(() -> {
+            try {
+                if (isLeader.get()) {
+                    long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
+                    long lastUpdateByLeader = assignmentPerister.getAssignmentsLastUpdatedByLeader();
+                    if (lastLocalUpdate > lastUpdateByLeader) {
+                        LOG.debug("Leader writing assignments lastLocalUpdate ({}) > lastUpdateByLeader ({})",
+                                new Date(lastLocalUpdate), new Date(lastUpdateByLeader));
+                        assignmentPerister.writeAssignments();
+                    } else {
+                        LOG.trace("Leader not writing assignments lastLocalUpdate ({}) <= lastUpdateByLeader ({})",
+                                new Date(lastLocalUpdate), new Date(lastUpdateByLeader));
+                    }
+                } else {
+                    assignmentPerister.readAssignments(true);
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }, balancerConfig.getPersistDelay(), balancerConfig.getPersistPeriod(), TimeUnit.MILLISECONDS);
     }
 
     private void readNonCachedMetricsIP() {
@@ -247,7 +249,7 @@ public class BalancedMetricResolver implements MetricResolver {
                 public void isLeader() {
                     LOG.info("this balancer is the leader");
                     isLeader.set(true);
-                    writeAssignmentsToHdfs();
+                    assignmentPerister.writeAssignments();
                 }
 
                 @Override
@@ -273,11 +275,11 @@ public class BalancedMetricResolver implements MetricResolver {
                 for (ServiceInstance<ServerDetails> si : instances) {
                     ServerDetails pl = si.getPayload();
                     TimelyBalancedHost tbh = TimelyBalancedHost.of(pl.getHost(), pl.getTcpPort(), pl.getHttpPort(),
-                            pl.getWsPort(), pl.getUdpPort());
+                            pl.getWsPort(), pl.getUdpPort(), new ArrivalRate(arrivalRateExecutor));
-                    LOG.info("adding service {} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", si.getId(),
+                    LOG.info("adding service {} host:{}", si.getId(), TimelyBalancedHost.toStringShort(tbh));
-                            pl.getHost(), pl.getTcpPort(), pl.getHttpPort(), pl.getWsPort(), pl.getUdpPort());
                     tbh.setBalancerConfig(balancerConfig);
                     serverList.add(tbh);
+                    metricHostEvent(tbh, MetricHostEvent.ActionType.ADDED);
                 }
                 healthChecker.setTimelyHosts(serverList);
             } finally {
@@ -298,7 +300,8 @@ public class BalancedMetricResolver implements MetricResolver {
                         for (ServiceInstance<ServerDetails> si : instances) {
                             ServerDetails pl = (ServerDetails) si.getPayload();
                             TimelyBalancedHost tbh = TimelyBalancedHost.of(pl.getHost(), pl.getTcpPort(),
-                                    pl.getHttpPort(), pl.getWsPort(), pl.getUdpPort());
+                                    pl.getHttpPort(), pl.getWsPort(), pl.getUdpPort(),
+                                    new ArrivalRate(arrivalRateExecutor));
                             tbh.setBalancerConfig(balancerConfig);
                             availableHosts.add(tbh);
                         }
@@ -312,9 +315,8 @@ public class BalancedMetricResolver implements MetricResolver {
                                 availableHosts.remove(h);
                             } else {
                                 itr.remove();
-                                LOG.info("removing service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}",
+                                metricHostEvent(h, MetricHostEvent.ActionType.REMOVED);
-                                        h.getHost(), h.getTcpPort(), h.getHost(), h.getTcpPort(), h.getHttpPort(),
+                                LOG.info("removing service {}", TimelyBalancedHost.toStringShort(h));
-                                        h.getWsPort(), h.getUdpPort());
                                 for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
                                     if (e.getValue().equals(h)) {
                                         reassignMetrics.add(e.getKey());
@@ -326,19 +328,22 @@ public class BalancedMetricResolver implements MetricResolver {
 
                         // add new hosts that were not previously known
                         for (TimelyBalancedHost h : availableHosts) {
-                            LOG.info("adding service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}",
+                            LOG.info("adding service {}", TimelyBalancedHost.toStringShort(h));
-                                    h.getHost(), h.getTcpPort(), h.getHost(), h.getTcpPort(), h.getHttpPort(),
-                                    h.getWsPort(), h.getUdpPort());
                             serverList.add(h);
+                            metricHostEvent(h, MetricHostEvent.ActionType.ADDED);
                             rebalanceNeeded = true;
                         }
                         healthChecker.setTimelyHosts(serverList);
-
+                        if (!availableHosts.isEmpty()) {
+                            reassignMetrics.addAll(unassignedMetrics);
+                            unassignedMetrics.clear();
+                        }
                         if (isLeader.get()) {
                             for (String s : reassignMetrics) {
                                 TimelyBalancedHost h = getRoundRobinHost(null);
-                                assignMetric(s, h);
+                                assignMetric(s, h, HOST_REMOVED);
-                                LOG.debug("Assigned server removed.  Assigning {} to server {}:{}", s, h.getTcpPort());
+                                LOG.debug("Assigned server removed.  Assigning {} to server {}", s,
+                                        TimelyBalancedHost.toStringShort(h));
                             }
                         }
 
@@ -349,7 +354,7 @@ public class BalancedMetricResolver implements MetricResolver {
                         if (rebalanceNeeded) {
                             balance();
                         }
-                        writeAssignmentsToHdfs();
+                        assignmentPerister.writeAssignments();
                     }
                 }
 
@@ -409,7 +414,7 @@ public class BalancedMetricResolver implements MetricResolver {
         return tbh;
     }
 
-    private TimelyBalancedHost getRoundRobinHost(TimelyBalancedHost notThisOne) {
+    protected TimelyBalancedHost getRoundRobinHost(TimelyBalancedHost notThisOne) {
         TimelyBalancedHost tbh = null;
         balancerLock.readLock().lock();
         try {
@@ -531,8 +536,10 @@ public class BalancedMetricResolver implements MetricResolver {
 
         int numReassigned = 0;
         if (isLeader.get()) {
+            metricBalanceEvent(MetricBalanceEvent.ProgressType.BEGIN, balanceType, 0);
             balancerLock.writeLock().lock();
             try {
+                metricBalanceEvent(MetricBalanceEvent.ProgressType.BALANCER_LOCK_ACQUIRED, balanceType, 0);
                 Map<String, ArrivalRate> tempMetricMap = new HashMap<>();
                 tempMetricMap.putAll(metricMap);
 
@@ -562,8 +569,8 @@ public class BalancedMetricResolver implements MetricResolver {
                         }
                     }
 
-                    LOG.trace("focusHost {}:{} desiredChange:{} rateSortedMetrics.size():{}", h.getHost(),
+                    LOG.trace("focusHost {} desiredChange:{} rateSortedMetrics.size():{}",
-                            h.getTcpPort(), desiredChange, rateSortedMetrics.size());
+                            TimelyBalancedHost.toStringShort(h), desiredChange, rateSortedMetrics.size());
 
                     boolean doneWithHost = false;
                     while (!doneWithHost) {
@@ -592,7 +599,7 @@ public class BalancedMetricResolver implements MetricResolver {
                                 } else {
                                     String metric = e.getValue();
                                     Double metricRate = e.getKey();
-                                    assignMetric(metric, candidateHost);
+                                    assignMetric(metric, candidateHost, REBALANCE);
                                     numReassigned++;
                                     calculatedRates.put(candidateHost, calculatedRates.get(candidateHost) + metricRate);
                                     calculatedRates.put(h, calculatedRates.get(h) - metricRate);
@@ -601,9 +608,9 @@ public class BalancedMetricResolver implements MetricResolver {
                                     itr.remove();
                                     tempMetricMap.remove(metric);
                                     LOG.info(
-                                            "rebalancing: reassigning metric:{} rate:{} from server {}:{} to {}:{} remaining delta {}",
+                                            "rebalancing: reassigning metric:{} rate:{} from server {} to {} remaining delta {}",
-                                            metric, metricRate, h.getHost(), h.getTcpPort(), candidateHost.getHost(),
+                                            metric, metricRate, TimelyBalancedHost.toStringShort(h),
-                                            candidateHost.getTcpPort(), desiredChange);
+                                            TimelyBalancedHost.toStringShort(candidateHost), desiredChange);
                                 }
                             }
                         }
@@ -626,6 +633,7 @@ public class BalancedMetricResolver implements MetricResolver {
                 }
             } finally {
                 balancerLock.writeLock().unlock();
+                metricBalanceEvent(MetricBalanceEvent.ProgressType.END, balanceType, numReassigned);
             }
         }
         return numReassigned;
@@ -698,7 +706,7 @@ public class BalancedMetricResolver implements MetricResolver {
         }
     }
 
-    private boolean shouldCache(String metricName) {
+    protected boolean shouldCache(String metricName) {
 
         if (StringUtils.isBlank(metricName)) {
             return false;
@@ -747,7 +755,7 @@ public class BalancedMetricResolver implements MetricResolver {
             if (chooseMetricSpecificHost) {
                 ArrivalRate rate = metricMap.get(metric);
                 if (rate == null) {
-                    rate = new ArrivalRate(arrivalRateTimer);
+                    rate = new ArrivalRate(arrivalRateExecutor);
                     if (!balancerLock.isWriteLockedByCurrentThread()) {
                         balancerLock.readLock().unlock();
                         balancerLock.writeLock().lock();
@@ -760,7 +768,7 @@ public class BalancedMetricResolver implements MetricResolver {
             }
 
             if (StringUtils.isBlank(metric)) {
-                tbh = getRandomHost(null);
+                tbh = getRoundRobinHost(null);
             } else {
                 tbh = metricToHostMap.get(metric);
                 if (tbh == null) {
@@ -769,17 +777,17 @@ public class BalancedMetricResolver implements MetricResolver {
                         balancerLock.readLock().unlock();
                         balancerLock.writeLock().lock();
                     }
-                    assignMetric(metric, tbh);
+                    assignMetric(metric, tbh, ASSIGN_ROUND_ROBIN);
                 } else if (!tbh.isUp()) {
                     TimelyBalancedHost oldTbh = tbh;
-                    tbh = getLeastUsedHost();
+                    tbh = getRoundRobinHost(oldTbh);
-                    LOG.debug("rebalancing from host that is down: reassigning metric {} from server {}:{} to {}:{}",
+                    LOG.debug("rebalancing from host that is down: reassigning metric {} from server {} to {}", metric,
-                            metric, oldTbh.getHost(), oldTbh.getTcpPort(), tbh.getHost(), tbh.getTcpPort());
+                            TimelyBalancedHost.toStringShort(oldTbh), TimelyBalancedHost.toStringShort(tbh));
                     if (!balancerLock.isWriteLockedByCurrentThread()) {
                         balancerLock.readLock().unlock();
                         balancerLock.writeLock().lock();
                     }
-                    assignMetric(metric, tbh);
+                    assignMetric(metric, tbh, HOST_DOWN_ROUND_ROBIN);
                 }
             }
 
@@ -796,7 +804,7 @@ public class BalancedMetricResolver implements MetricResolver {
                         balancerLock.readLock().unlock();
                         balancerLock.writeLock().lock();
                     }
-                    assignMetric(metric, tbh);
+                    assignMetric(metric, tbh, ASSIGN_FALLBACK_SEQUENTIAL);
                 }
             }
         } finally {
@@ -825,20 +833,23 @@ public class BalancedMetricResolver implements MetricResolver {
                 metric = null;
             }
 
-            if (tbh == null || !tbh.isUp()) {
+            if (tbh == null) {
-                for (int x = 0; tbh == null && x < serverList.size(); x++) {
+                tbh = getRoundRobinHost(null);
-                    tbh = serverList.get(Math.abs(r.nextInt() & Integer.MAX_VALUE) % serverList.size());
+                if (!balancerLock.isWriteLockedByCurrentThread()) {
-                    if (!tbh.isUp()) {
+                    balancerLock.readLock().unlock();
-                        tbh = null;
+                    balancerLock.writeLock().lock();
-                    }
                 }
-                if (tbh != null && StringUtils.isNotBlank(metric)) {
+                assignMetric(metric, tbh, ASSIGN_ROUND_ROBIN);
-                    if (!balancerLock.isWriteLockedByCurrentThread()) {
+            } else if (!tbh.isUp()) {
-                        balancerLock.readLock().unlock();
+                TimelyBalancedHost oldTbh = tbh;
-                        balancerLock.writeLock().lock();
+                tbh = getRoundRobinHost(oldTbh);
-                    }
+                LOG.debug("rebalancing from host that is down: reassigning metric {} from server {} to {}", metric,
-                    assignMetric(metric, tbh);
+                        TimelyBalancedHost.toStringShort(oldTbh), TimelyBalancedHost.toStringShort(tbh));
+                if (!balancerLock.isWriteLockedByCurrentThread()) {
+                    balancerLock.readLock().unlock();
+                    balancerLock.writeLock().lock();
                 }
+                assignMetric(metric, tbh, HOST_DOWN_ROUND_ROBIN);
             }
 
             // if all else fails
@@ -854,7 +865,7 @@ public class BalancedMetricResolver implements MetricResolver {
                         balancerLock.readLock().unlock();
                         balancerLock.writeLock().lock();
                     }
-                    assignMetric(metric, tbh);
+                    assignMetric(metric, tbh, ASSIGN_FALLBACK_SEQUENTIAL);
                 }
             }
         } finally {
@@ -867,7 +878,7 @@ public class BalancedMetricResolver implements MetricResolver {
         return tbh;
     }
 
-    private TimelyBalancedHost findHost(String host, int tcpPort) {
+    protected TimelyBalancedHost findHost(String host, int tcpPort) {
         TimelyBalancedHost tbh = null;
         for (TimelyBalancedHost h : serverList) {
             if (h.getHost().equals(host) && h.getTcpPort() == tcpPort) {
@@ -878,179 +889,100 @@ public class BalancedMetricResolver implements MetricResolver {
         return tbh;
     }
 
-    private void readAssignmentsFromHdfs(boolean checkIfNecessary) {
+    private void assignMetric(String metric, TimelyBalancedHost tbh, MetricAssignedEvent.Reason reason) {
-
-        try {
-            long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
-            long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
-            if (checkIfNecessary) {
-                if (lastHdfsUpdate <= lastLocalUpdate) {
-                    LOG.debug("Not reading assignments from hdfs lastHdfsUpdate ({}) <= lastLocalUpdate ({})",
-                            new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
-                    return;
-                }
-            }
-            LOG.debug("Reading assignments from hdfs lastHdfsUpdate ({}) > lastLocalUpdate ({})",
-                    new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
-        } catch (Exception e) {
-            LOG.error(e.getMessage(), e);
-        }
-
-        // proceed with reading from HDFS
-
-        Map<String, TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
-        try {
-            boolean acquired = false;
-            while (!acquired) {
-                acquired = assignmentsIPRWLock.readLock().acquire(60, TimeUnit.SECONDS);
-            }
-            balancerLock.writeLock().lock();
-            Configuration configuration = new Configuration();
-            FileSystem fs = FileSystem.get(configuration);
-            Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
-            FSDataInputStream iStream = fs.open(assignmentFile);
-
-            CsvReader reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
-            reader.setUseTextQualifier(false);
-
-            // skip the headers
-            boolean success = true;
-            success = reader.readHeaders();
-
-            while (success) {
-                success = reader.readRecord();
-                String[] nextLine = reader.getValues();
-                if (nextLine.length >= 3) {
-                    String metric = nextLine[0];
-                    String host = nextLine[1];
-                    int tcpPort = Integer.parseInt(nextLine[2]);
-                    TimelyBalancedHost tbh = findHost(host, tcpPort);
-                    if (tbh == null) {
-                        tbh = getRoundRobinHost(null);
-                    } else {
-                        LOG.trace("Found assigment: {} to {}:{}", metric, host, tcpPort);
-                    }
-                    assignedMetricToHostMap.put(metric, tbh);
-                }
-            }
-
-            metricToHostMap.clear();
-            for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
-                if (StringUtils.isNotBlank(e.getKey()) && e.getValue() != null) {
-                    if (shouldCache(e.getKey())) {
-                        metricToHostMap.put(e.getKey(), e.getValue());
-                    }
-                } else {
-                    Exception e1 = new IllegalStateException(
-                            "Bad assignment metric:" + e.getKey() + " host:" + e.getValue());
-                    LOG.warn(e1.getMessage(), e1);
-                }
-            }
-
-            nonCachedMetricsLocalLock.readLock().lock();
-            try {
-                Iterator<Map.Entry<String, TimelyBalancedHost>> itr = metricToHostMap.entrySet().iterator();
-                while (itr.hasNext()) {
-                    Map.Entry<String, TimelyBalancedHost> e = itr.next();
-                    if (nonCachedMetrics.contains(e.getKey())) {
-                        itr.remove();
-                    }
-                }
-            } finally {
-                nonCachedMetricsLocalLock.readLock().unlock();
-            }
-            assignmentsLastUpdatedLocal.set(assignmentsLastUpdatedInHdfs.get().postValue());
-            LOG.info("Read {} assignments from hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
-                    new Date(assignmentsLastUpdatedLocal.get()));
-        } catch (Exception e) {
-            LOG.error(e.getMessage(), e);
-        } finally {
-            balancerLock.writeLock().unlock();
-            try {
-                assignmentsIPRWLock.readLock().release();
-            } catch (Exception e) {
-                LOG.error(e.getMessage(), e);
-            }
-        }
-    }
-
-    private void writeAssignmentsToHdfs() {
-
-        CsvWriter writer = null;
-        try {
-            boolean acquired = false;
-            while (!acquired) {
-                acquired = assignmentsIPRWLock.writeLock().acquire(60, TimeUnit.SECONDS);
-            }
-            balancerLock.readLock().lock();
-            nonCachedMetricsLocalLock.readLock().lock();
-            try {
-                if (!metricToHostMap.isEmpty()) {
-                    Configuration configuration = new Configuration();
-                    FileSystem fs = FileSystem.get(configuration);
-                    Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
-                    if (!fs.exists(assignmentFile.getParent())) {
-                        fs.mkdirs(assignmentFile.getParent());
-                    }
-                    FSDataOutputStream oStream = fs.create(assignmentFile, true);
-                    writer = new CsvWriter(oStream, ',', Charset.forName("UTF-8"));
-                    writer.setUseTextQualifier(false);
-                    writer.write("metric");
-                    writer.write("host");
-                    writer.write("tcpPort");
-                    writer.endRecord();
-                    for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
-                        if (!nonCachedMetrics.contains(e.getKey())) {
-                            writer.write(e.getKey());
-                            writer.write(e.getValue().getHost());
-                            writer.write(Integer.toString(e.getValue().getTcpPort()));
-                            writer.endRecord();
-                            LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(),
-                                    e.getValue().getTcpPort());
-                        }
-                    }
-
-                    long now = System.currentTimeMillis();
-                    assignmentsLastUpdatedLocal.set(now);
-                    assignmentsLastUpdatedInHdfs.trySet(now);
-                    if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
-                        assignmentsLastUpdatedInHdfs.forceSet(now);
-                    }
-                    LOG.debug("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})",
-                            metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
-                }
-            } finally {
-                nonCachedMetricsLocalLock.readLock().unlock();
-                balancerLock.readLock().unlock();
-            }
-        } catch (Exception e) {
-            LOG.error(e.getMessage(), e);
-        } finally {
-            if (writer != null) {
-                writer.close();
-            }
-            try {
-                assignmentsIPRWLock.writeLock().release();
-            } catch (Exception e) {
-                LOG.error(e.getMessage(), e);
-            }
-        }
-    }
-
-    private void assignMetric(String metric, TimelyBalancedHost tbh) {
         if (isLeader.get()) {
             if (StringUtils.isNotBlank(metric) && tbh != null) {
                 balancerLock.writeLock().lock();
                 try {
+                    metricAssignedEvent(metric, metricToHostMap.get(metric), tbh, reason);
                     metricToHostMap.put(metric, tbh);
                     assignmentsLastUpdatedLocal.set(System.currentTimeMillis());
                 } finally {
                     balancerLock.writeLock().unlock();
                 }
             } else {
-                Exception e = new IllegalStateException("Bad assignment metric:" + metric + " host:" + tbh);
+                unassignedMetrics.add(metric);
-                LOG.warn(e.getMessage(), e);
+                // Exception e = new IllegalStateException("Bad assignment metric:" + metric + "
+                // host:" + tbh);
+                LOG.warn("Bad assignment metric:" + metric + " host:" + tbh);
             }
         }
     }
+
+    public void registerCallback(MetricAssignedCallback callback) {
+        this.metricAssignedCallbacks.add(callback);
+    }
+
+    public void registerCallback(MetricBalanceCallback callback) {
+        this.metricBalanceCallbacks.add(callback);
+    }
+
+    public void registerCallback(MetricHostCallback callback) {
+        this.metricHostCallbacks.add(callback);
+    }
+
+    protected void metricAssignedEvent(String metric, TimelyBalancedHost losing, TimelyBalancedHost gaining,
+            MetricAssignedEvent.Reason reason) {
+        if (!metricAssignedCallbacks.isEmpty()) {
+            MetricAssignedEvent event = new MetricAssignedEvent(metric, losing, gaining, reason);
+            for (MetricAssignedCallback callback : metricAssignedCallbacks) {
+                callback.onEvent(event);
+            }
+        }
+    }
+
+    protected void metricBalanceEvent(MetricBalanceEvent.ProgressType progressType,
+            BalancedMetricResolver.BalanceType balanceType, long numReassigned) {
+        if (!metricBalanceCallbacks.isEmpty()) {
+            MetricBalanceEvent event = new MetricBalanceEvent(progressType, balanceType, numReassigned);
+            for (MetricBalanceCallback callback : metricBalanceCallbacks) {
+                callback.onEvent(event);
+            }
+        }
+    }
+
+    protected void metricHostEvent(TimelyBalancedHost host, MetricHostEvent.ActionType actionType) {
+        if (!metricHostCallbacks.isEmpty()) {
+            MetricHostEvent event = new MetricHostEvent(host, actionType);
+            for (MetricHostCallback callback : metricHostCallbacks) {
+                callback.onEvent(event);
+            }
+        }
+    }
+
+    protected void writeAssignments(Map<String, TimelyBalancedHost> metricToHostMap) {
+        assignmentPerister.writeAssignments(metricToHostMap);
+    }
+
+    @Override
+    public void stop() {
+
+        this.assignmentExecutor.shutdown();
+        try {
+            this.assignmentExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.assignmentExecutor.isTerminated()) {
+                this.assignmentExecutor.shutdownNow();
+            }
+        }
+
+        this.arrivalRateExecutor.shutdown();
+        try {
+            this.arrivalRateExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.arrivalRateExecutor.isTerminated()) {
+                this.arrivalRateExecutor.shutdownNow();
+            }
+        }
+
+        try {
+            this.healthChecker.close();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+    }
 }

balancer/src/main/java/timely/balancer/resolver/MetricAssignmentPerister.java

@@ -0,0 +1,320 @@
+package timely.balancer.resolver;
+
+import static timely.balancer.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.Charset;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.csvreader.CsvReader;
+import com.csvreader.CsvWriter;
+import org.apache.commons.lang.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
+import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
+import org.apache.curator.retry.RetryForever;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.balancer.configuration.BalancerConfiguration;
+import timely.balancer.connection.TimelyBalancedHost;
+import timely.balancer.resolver.eventing.MetricAssignedEvent;
+
+public class MetricAssignmentPerister {
+
+    private final Logger LOG = LoggerFactory.getLogger(getClass());
+
+    private enum PeristerType {
+        HDFS, FILE
+    }
+
+    // passed in through constructor
+    BalancedMetricResolver balancedMetricResolver;
+    private BalancerConfiguration balancerConfig;
+    private Map<String, TimelyBalancedHost> metricToHostMap;
+    private Set<String> nonCachedMetrics;
+    private InterProcessReadWriteLock assignmentsIPRWLock;
+    private ReentrantReadWriteLock nonCachedMetricsLocalLock;
+    private ReentrantReadWriteLock balancerLock;
+    private AtomicLong assignmentsLastUpdatedLocal;
+
+    // local
+    private DistributedAtomicLong assignmentsLastUpdatedInHdfs;
+
+    public MetricAssignmentPerister(BalancedMetricResolver balancedMetricResolver, BalancerConfiguration balancerConfig,
+            Map<String, TimelyBalancedHost> metricToHostMap, Set<String> nonCachedMetrics,
+            CuratorFramework curatorFramework, InterProcessReadWriteLock assignmentsIPRWLock,
+            ReentrantReadWriteLock nonCachedMetricsLocalLock, AtomicLong assignmentsLastUpdatedLocal,
+            ReentrantReadWriteLock balancerLock) {
+
+        this.balancedMetricResolver = balancedMetricResolver;
+        this.balancerConfig = balancerConfig;
+        this.metricToHostMap = metricToHostMap;
+        this.nonCachedMetrics = nonCachedMetrics;
+        this.assignmentsIPRWLock = assignmentsIPRWLock;
+        this.nonCachedMetricsLocalLock = nonCachedMetricsLocalLock;
+        this.assignmentsLastUpdatedLocal = assignmentsLastUpdatedLocal;
+        this.balancerLock = balancerLock;
+        assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH,
+                new RetryForever(1000));
+    }
+
+    public long getAssignmentsLastUpdatedByLeader() throws Exception {
+        return assignmentsLastUpdatedInHdfs.get().postValue();
+    }
+
+    public void readAssignments(boolean checkIfNecessary) {
+
+        try {
+            long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
+            long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
+            if (checkIfNecessary) {
+                if (lastHdfsUpdate <= lastLocalUpdate) {
+                    LOG.debug("Not reading assignments from storage lastStorageUpdate ({}) <= lastLocalUpdate ({})",
+                            new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
+                    return;
+                }
+            }
+            LOG.debug("Reading assignments from storage lastStorageUpdate ({}) > lastLocalUpdate ({})",
+                    new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+
+        // proceed with reading from storage
+
+        try (InputStream inputStream = getInputStream(balancerConfig)) {
+            boolean acquired = false;
+            while (!acquired) {
+                acquired = assignmentsIPRWLock.readLock().acquire(60, TimeUnit.SECONDS);
+            }
+            balancerLock.writeLock().lock();
+            // get potential new list from implementation
+            Map<String, TimelyBalancedHost> assignedMetricToHostMap = readFromStream(inputStream);
+            metricToHostMap.clear();
+            for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
+                if (StringUtils.isNotBlank(e.getKey()) && e.getValue() != null) {
+                    if (balancedMetricResolver.shouldCache(e.getKey())) {
+                        balancedMetricResolver.metricAssignedEvent(e.getKey(), null, e.getValue(),
+                                MetricAssignedEvent.Reason.ASSIGN_FILE);
+                        metricToHostMap.put(e.getKey(), e.getValue());
+                    }
+                } else {
+                    Exception e1 = new IllegalStateException(
+                            "Bad assignment metric:" + e.getKey() + " host:" + e.getValue());
+                    LOG.warn(e1.getMessage(), e1);
+                }
+            }
+            nonCachedMetricsLocalLock.readLock().lock();
+            try {
+                Iterator<Map.Entry<String, TimelyBalancedHost>> itr = metricToHostMap.entrySet().iterator();
+                while (itr.hasNext()) {
+                    Map.Entry<String, TimelyBalancedHost> e = itr.next();
+                    if (nonCachedMetrics.contains(e.getKey())) {
+                        itr.remove();
+                    }
+                }
+            } finally {
+                nonCachedMetricsLocalLock.readLock().unlock();
+            }
+            assignmentsLastUpdatedLocal.set(assignmentsLastUpdatedInHdfs.get().postValue());
+            LOG.info("Read {} assignments from storage lastStorageUpdate = lastLocalUpdate ({})",
+                    metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
+        } catch (FileNotFoundException e) {
+            createAssignmentFile();
+            metricToHostMap.clear();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            balancerLock.writeLock().unlock();
+            try {
+                assignmentsIPRWLock.readLock().release();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+    }
+
+    public void writeAssignments() {
+        writeAssignments(metricToHostMap);
+    }
+
+    protected void writeAssignments(Map<String, TimelyBalancedHost> metricToHostMap) {
+
+        try {
+            boolean acquired = false;
+            while (!acquired) {
+                acquired = assignmentsIPRWLock.writeLock().acquire(60, TimeUnit.SECONDS);
+            }
+            balancerLock.readLock().lock();
+            nonCachedMetricsLocalLock.readLock().lock();
+            try (OutputStream outputStream = getOutputStream(balancerConfig)) {
+                if (!metricToHostMap.isEmpty()) {
+                    writeToStream(outputStream);
+                    long now = System.currentTimeMillis();
+                    assignmentsLastUpdatedLocal.set(now);
+                    assignmentsLastUpdatedInHdfs.trySet(now);
+                    if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
+                        assignmentsLastUpdatedInHdfs.forceSet(now);
+                    }
+                    LOG.debug("Wrote {} assignments to storage lastStorageUpdate = lastLocalUpdate ({})",
+                            metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
+                }
+            } finally {
+                nonCachedMetricsLocalLock.readLock().unlock();
+                balancerLock.readLock().unlock();
+            }
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            try {
+                assignmentsIPRWLock.writeLock().release();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+    }
+
+    public Map<String, TimelyBalancedHost> readFromStream(InputStream inputStream) throws IOException {
+        Map<String, TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
+        try {
+            CsvReader reader = new CsvReader(inputStream, ',', Charset.forName("UTF-8"));
+            reader.setUseTextQualifier(false);
+
+            // skip the headers
+            boolean success = true;
+            success = reader.readHeaders();
+
+            while (success) {
+                success = reader.readRecord();
+                String[] nextLine = reader.getValues();
+                if (nextLine.length >= 3) {
+                    String metric = nextLine[0];
+                    String host = nextLine[1];
+                    int tcpPort = Integer.parseInt(nextLine[2]);
+                    TimelyBalancedHost tbh = balancedMetricResolver.findHost(host, tcpPort);
+                    if (tbh == null) {
+                        tbh = balancedMetricResolver.getRoundRobinHost(null);
+                    } else {
+                        LOG.trace("Found assigment: {} to {}:{}", metric, host, tcpPort);
+                    }
+                    assignedMetricToHostMap.put(metric, tbh);
+                }
+            }
+        } catch (IOException e) {
+            LOG.error(e.getMessage(), e);
+            throw e;
+        }
+        return assignedMetricToHostMap;
+    }
+
+    private void writeToStream(OutputStream outputStream) throws IOException {
+
+        CsvWriter writer = null;
+        try {
+            writer = new CsvWriter(outputStream, ',', Charset.forName("UTF-8"));
+            writer.setUseTextQualifier(false);
+            writer.write("metric");
+            writer.write("host");
+            writer.write("tcpPort");
+            writer.endRecord();
+            for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
+                if (!nonCachedMetrics.contains(e.getKey())) {
+                    writer.write(e.getKey());
+                    writer.write(e.getValue().getHost());
+                    writer.write(Integer.toString(e.getValue().getTcpPort()));
+                    writer.endRecord();
+                    LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(),
+                            e.getValue().getTcpPort());
+                }
+            }
+        } finally {
+            if (writer != null) {
+                writer.close();
+            }
+        }
+    }
+
+    public void createAssignmentFile() {
+        LOG.info("Creating assignment file: " + balancerConfig.getAssignmentFile());
+        PeristerType peristerType = PeristerType.valueOf(balancerConfig.getMetricAssignmentPersisterType());
+        switch (peristerType) {
+            case HDFS:
+                try {
+                    Configuration configuration = new Configuration();
+                    FileSystem fs = FileSystem.get(configuration);
+                    Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
+                    fs.create(assignmentFile);
+                } catch (Exception e) {
+                    LOG.error(e.getMessage(), e);
+                }
+                break;
+            case FILE:
+                try {
+                    File assignmentFile = new File(balancerConfig.getAssignmentFile());
+                    String parentPath = assignmentFile.getParent();
+                    File directory = new File(parentPath);
+                    if (!directory.exists()) {
+
+                        directory.mkdirs();
+                    }
+                    if (!assignmentFile.exists()) {
+                        assignmentFile.createNewFile();
+                    }
+                } catch (Exception e) {
+                    LOG.error(e.getMessage(), e);
+                }
+                break;
+        }
+    }
+
+    public InputStream getInputStream(BalancerConfiguration balancerConfiguration) throws Exception {
+        PeristerType peristerType = PeristerType.valueOf(balancerConfiguration.getMetricAssignmentPersisterType());
+        switch (peristerType) {
+            case HDFS:
+                Configuration configuration = new Configuration();
+                FileSystem fs = FileSystem.get(configuration);
+                Path assignmentFile = new Path(balancerConfiguration.getAssignmentFile());
+                return fs.open(assignmentFile);
+            case FILE:
+                return new FileInputStream(new File(balancerConfiguration.getAssignmentFile()));
+            default:
+                throw new IllegalArgumentException(
+                        "Unknown metricPersisterType " + balancerConfiguration.getMetricAssignmentPersisterType());
+        }
+    }
+
+    public OutputStream getOutputStream(BalancerConfiguration balancerConfiguration) throws IOException {
+        PeristerType peristerType = PeristerType.valueOf(balancerConfiguration.getMetricAssignmentPersisterType());
+        switch (peristerType) {
+            case HDFS:
+                Configuration configuration = new Configuration();
+                FileSystem fs = FileSystem.get(configuration);
+                Path assignmentFile = new Path(balancerConfiguration.getAssignmentFile());
+                if (!fs.exists(assignmentFile.getParent())) {
+                    fs.mkdirs(assignmentFile.getParent());
+                }
+                return fs.create(assignmentFile, true);
+            case FILE:
+                return new FileOutputStream(new File(balancerConfiguration.getAssignmentFile()));
+            default:
+                throw new IllegalArgumentException(
+                        "Unknown metricPersisterType " + balancerConfiguration.getMetricAssignmentPersisterType());
+        }
+    }
+}

balancer/src/main/java/timely/balancer/resolver/MetricResolver.java

@@ -7,4 +7,9 @@ public interface MetricResolver {
     public TimelyBalancedHost getHostPortKeyIngest(String metric);
 
     public TimelyBalancedHost getHostPortKey(String metric);
+
+    public void start();
+
+    public void stop();
+
 }

balancer/src/main/java/timely/balancer/resolver/eventing/MetricAssignedCallback.java

@@ -0,0 +1,7 @@
+package timely.balancer.resolver.eventing;
+
+public abstract class MetricAssignedCallback {
+
+    public abstract void onEvent(MetricAssignedEvent event);
+
+}

balancer/src/main/java/timely/balancer/resolver/eventing/MetricAssignedEvent.java

@@ -0,0 +1,39 @@
+package timely.balancer.resolver.eventing;
+
+import timely.balancer.connection.TimelyBalancedHost;
+
+public class MetricAssignedEvent {
+
+    public enum Reason {
+        ASSIGN_ROUND_ROBIN, HOST_DOWN_ROUND_ROBIN, ASSIGN_FALLBACK_SEQUENTIAL, HOST_REMOVED, REBALANCE, ASSIGN_FILE
+    }
+
+    private final String metric;
+    private final TimelyBalancedHost losingHost;
+    private final TimelyBalancedHost gainingHost;
+    private final Reason reason;
+
+    public MetricAssignedEvent(String metric, TimelyBalancedHost losingHost, TimelyBalancedHost gainingHost,
+            Reason reason) {
+        this.metric = metric;
+        this.losingHost = losingHost;
+        this.gainingHost = gainingHost;
+        this.reason = reason;
+    }
+
+    public String getMetric() {
+        return metric;
+    }
+
+    public TimelyBalancedHost getLosingHost() {
+        return losingHost;
+    }
+
+    public TimelyBalancedHost getGainingHost() {
+        return gainingHost;
+    }
+
+    public Reason getReason() {
+        return reason;
+    }
+}

balancer/src/main/java/timely/balancer/resolver/eventing/MetricBalanceCallback.java

@@ -0,0 +1,7 @@
+package timely.balancer.resolver.eventing;
+
+public abstract class MetricBalanceCallback {
+
+    public abstract void onEvent(MetricBalanceEvent event);
+
+}

balancer/src/main/java/timely/balancer/resolver/eventing/MetricBalanceEvent.java

@@ -0,0 +1,33 @@
+package timely.balancer.resolver.eventing;
+
+import timely.balancer.resolver.BalancedMetricResolver;
+
+public class MetricBalanceEvent {
+
+    public enum ProgressType {
+        BEGIN, BALANCER_LOCK_ACQUIRED, END
+    }
+
+    private final ProgressType progressType;
+    private final BalancedMetricResolver.BalanceType balanceType;
+    private final Long numRessigned;
+
+    public MetricBalanceEvent(ProgressType progressType, BalancedMetricResolver.BalanceType balanceType,
+            long numReassigned) {
+        this.progressType = progressType;
+        this.balanceType = balanceType;
+        this.numRessigned = numReassigned;
+    }
+
+    public BalancedMetricResolver.BalanceType getBalanceType() {
+        return balanceType;
+    }
+
+    public ProgressType getProgressType() {
+        return progressType;
+    }
+
+    public Long getNumRessigned() {
+        return numRessigned;
+    }
+}

balancer/src/main/java/timely/balancer/resolver/eventing/MetricHostCallback.java

@@ -0,0 +1,7 @@
+package timely.balancer.resolver.eventing;
+
+public abstract class MetricHostCallback {
+
+    public abstract void onEvent(MetricHostEvent event);
+
+}

balancer/src/main/java/timely/balancer/resolver/eventing/MetricHostEvent.java

@@ -0,0 +1,26 @@
+package timely.balancer.resolver.eventing;
+
+import timely.balancer.connection.TimelyBalancedHost;
+
+public class MetricHostEvent {
+
+    public enum ActionType {
+        ADDED, REMOVED, SUCCESS, FAILURE
+    }
+
+    private final TimelyBalancedHost timelyBalancedHost;
+    private final ActionType actionType;
+
+    public MetricHostEvent(TimelyBalancedHost timelyBalancedHost, ActionType actionType) {
+        this.timelyBalancedHost = timelyBalancedHost;
+        this.actionType = actionType;
+    }
+
+    public TimelyBalancedHost getTimelyBalancedHost() {
+        return timelyBalancedHost;
+    }
+
+    public ActionType getActionType() {
+        return actionType;
+    }
+}

balancer/src/main/spotbugs/excludes.xml

@@ -10,4 +10,9 @@
     <Method name="getBuf" />
     <Bug pattern="EI_EXPOSE_REP" />
   </Match>
+  <Match>
+    <Class name="timely.balancer.resolver.MetricAssignmentPerister" />
+    <Method name="createAssignmentFile" />
+    <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />
+  </Match>
 </FindBugsFilter>

balancer/src/test/java/timely/balancer/resolver/ServiceRegistry.java

@@ -0,0 +1,76 @@
+package timely.balancer.resolver;
+
+import static timely.Server.SERVICE_DISCOVERY_PATH;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.x.discovery.ServiceDiscovery;
+import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
+import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.curator.x.discovery.ServiceInstanceBuilder;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.ServerDetails;
+import timely.balancer.connection.TimelyBalancedHost;
+
+public class ServiceRegistry {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ServiceRegistry.class);
+
+    public static void registerService(CuratorFramework curatorFramework, TimelyBalancedHost tbh) {
+        try {
+            try {
+                Stat stat = curatorFramework.checkExists().forPath(SERVICE_DISCOVERY_PATH);
+                if (stat == null) {
+                    curatorFramework.create().creatingParentContainersIfNeeded().forPath(SERVICE_DISCOVERY_PATH);
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage());
+            }
+
+            ServiceInstance<ServerDetails> serviceInstance = serviceInstance(tbh);
+            ServiceDiscovery<ServerDetails> discovery = ServiceDiscoveryBuilder.builder(ServerDetails.class)
+                    .client(curatorFramework).basePath(SERVICE_DISCOVERY_PATH).build();
+            discovery.start();
+            discovery.registerService(serviceInstance);
+        } catch (Exception e) {
+            LOG.error(e.getMessage());
+        }
+    }
+
+    public static void unregisterService(CuratorFramework curatorFramework, TimelyBalancedHost tbh) {
+        try {
+            try {
+                Stat stat = curatorFramework.checkExists().forPath(SERVICE_DISCOVERY_PATH);
+                if (stat == null) {
+                    curatorFramework.create().creatingParentContainersIfNeeded().forPath(SERVICE_DISCOVERY_PATH);
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage());
+            }
+
+            ServiceInstance<ServerDetails> serviceInstance = serviceInstance(tbh);
+            ServiceDiscovery<ServerDetails> discovery = ServiceDiscoveryBuilder.builder(ServerDetails.class)
+                    .client(curatorFramework).basePath(SERVICE_DISCOVERY_PATH).build();
+            discovery.start();
+            discovery.unregisterService(serviceInstance);
+        } catch (Exception e) {
+            LOG.error(e.getMessage());
+        }
+    }
+
+    private static ServiceInstance<ServerDetails> serviceInstance(TimelyBalancedHost config) throws Exception {
+        ServerDetails serverDetails = new ServerDetails();
+        serverDetails.setHost(config.getHost());
+        serverDetails.setTcpPort(config.getTcpPort());
+        serverDetails.setHttpPort(config.getHttpPort());
+        serverDetails.setWsPort(config.getWsPort());
+        serverDetails.setUdpPort(config.getUdpPort());
+
+        ServiceInstanceBuilder<ServerDetails> builder = ServiceInstance.builder();
+        String serviceName = config.getHost() + ":" + config.getTcpPort();
+        ServiceInstance<ServerDetails> serviceInstance = builder.id(serviceName).name("timely-server")
+                .address(config.getHost()).port(config.getTcpPort()).payload(serverDetails).build();
+        return serviceInstance;
+    }
+}

balancer/src/test/java/timely/balancer/resolver/TestBalancedMetricResolver.java

@@ -0,0 +1,270 @@
+package timely.balancer.resolver;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.UUID;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServerMain;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.SocketUtils;
+import timely.balancer.configuration.BalancerConfiguration;
+import timely.balancer.connection.TimelyBalancedHost;
+import timely.balancer.resolver.eventing.MetricAssignedCallback;
+import timely.balancer.resolver.eventing.MetricAssignedEvent;
+import timely.balancer.resolver.eventing.MetricBalanceCallback;
+import timely.balancer.resolver.eventing.MetricBalanceEvent;
+import timely.balancer.resolver.eventing.MetricHostCallback;
+import timely.balancer.resolver.eventing.MetricHostEvent;
+
+public class TestBalancedMetricResolver {
+
+    private final Logger LOG = LoggerFactory.getLogger(getClass());
+    private CuratorFramework curatorFramework;
+    private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HHmmss.SSS");
+    private Random r = new Random(System.currentTimeMillis());
+    private List<TimelyBalancedHost> timelyHosts;
+    private Map<MetricAssignedEvent, Integer> metricBalanceEventCounter = new HashMap<>();
+
+    @Before
+    public void startup() {
+        QuorumPeerConfig currentConfig = createZookeeper();
+        InetSocketAddress address = currentConfig.getClientPortAddress();
+        String zkServer = address.getHostName() + ":" + address.getPort();
+        curatorFramework = CuratorFrameworkFactory.newClient(zkServer, 1000, 5000, new RetryNTimes(2, 50));
+        curatorFramework.start();
+        timelyHosts = new ArrayList<>();
+    }
+
+    @After
+    public void shutdown() {
+        curatorFramework.close();
+    }
+
+    private QuorumPeerConfig createZookeeper() {
+        Properties startupProperties = new Properties();
+        File file = new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID());
+        file.deleteOnExit();
+        startupProperties.setProperty("dataDir", file.getAbsolutePath());
+        startupProperties.setProperty("clientPort", String.valueOf(SocketUtils.findAvailableTcpPort(3000, 4000)));
+        QuorumPeerConfig quorumConfiguration = new QuorumPeerConfig();
+
+        try {
+            quorumConfiguration.parseProperties(startupProperties);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+
+        ZooKeeperServerMain zooKeeperServer = new ZooKeeperServerMain();
+        final ServerConfig configuration = new ServerConfig();
+        configuration.readFrom(quorumConfiguration);
+
+        new Thread() {
+
+            public void run() {
+                try {
+                    zooKeeperServer.runFromConfig(configuration);
+                } catch (IOException e) {
+                    LOG.error("ZooKeeper Failed", e);
+                }
+            }
+        }.start();
+
+        return quorumConfiguration;
+    }
+
+    private BalancerConfiguration getBalancerConfiguration() {
+        BalancerConfiguration balancerConfiguration = new BalancerConfiguration();
+        balancerConfiguration.setMetricAssignmentPersisterType("FILE");
+        File file = new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID());
+        file.deleteOnExit();
+        balancerConfiguration.setAssignmentFile(file.getAbsolutePath());
+        balancerConfiguration.setBalanceDelay(0);
+        balancerConfiguration.setBalancePeriod(1000);
+        balancerConfiguration.setPersistDelay(0);
+        balancerConfiguration.setPersistDelay(1000);
+        balancerConfiguration.setCheckServerHealthInterval(10);
+        return balancerConfiguration;
+    }
+
+    private void createTimelyHosts(BalancerConfiguration balancerConfiguration, int numHosts) {
+        for (int port = 1000, x = 0; x < numHosts; x++, port += 10) {
+            TimelyBalancedHost tbh = TimelyBalancedHost.of("localhost", port);
+            tbh.setBalancerConfig(balancerConfiguration);
+            timelyHosts.add(tbh);
+            ServiceRegistry.registerService(curatorFramework, tbh);
+        }
+    }
+
+    private void registerCallbacks(BalancedMetricResolver resolver) {
+        resolver.registerCallback(new MetricAssignedCallback() {
+
+            @Override
+            public void onEvent(MetricAssignedEvent event) {
+                LOG.debug(String.format("%s metric:%s lose:%s:%d gain:%s:%d reason:%s",
+                        event.getClass().getSimpleName(), event.getMetric(),
+                        event.getLosingHost() == null ? null : event.getLosingHost().getHost(),
+                        event.getLosingHost() == null ? null : event.getLosingHost().getTcpPort(),
+                        event.getGainingHost().getHost(), event.getGainingHost().getTcpPort(), event.getReason()));
+            }
+        });
+
+        resolver.registerCallback(new MetricHostCallback() {
+
+            @Override
+            public void onEvent(MetricHostEvent event) {
+                LOG.debug(String.format("%s host:%s:%d type:%s", event.getClass().getSimpleName(),
+                        event.getTimelyBalancedHost().getHost(), event.getTimelyBalancedHost().getTcpPort(),
+                        event.getActionType()));
+            }
+        });
+
+        resolver.registerCallback(new MetricBalanceCallback() {
+
+            @Override
+            public void onEvent(MetricBalanceEvent event) {
+                LOG.debug(String.format("%s type:%s progress:%s reassigned:%d", event.getClass().getSimpleName(),
+                        event.getBalanceType(), event.getProgressType(), event.getNumRessigned()));
+            }
+        });
+    }
+
+    private TimelyBalancedHost randomServerDown(TestHealthChecker healthChecker, int numHosts) {
+        TimelyBalancedHost tbh = timelyHosts.get(r.nextInt(numHosts));
+        LOG.debug("Marking server down:" + TimelyBalancedHost.toStringShort(tbh));
+        healthChecker.serverDown(tbh);
+        return tbh;
+    }
+
+    private TimelyBalancedHost randomServerToggle(TestHealthChecker healthChecker, int numHosts) {
+        TimelyBalancedHost tbh = timelyHosts.get(r.nextInt(numHosts));
+        LOG.debug("Marking server " + (tbh.isUp() ? "down:" : "up:") + TimelyBalancedHost.toStringShort(tbh));
+        healthChecker.serverToggle(tbh);
+        return tbh;
+    }
+
+    private TimelyBalancedHost randomServerUnregister(int numHosts) {
+        TimelyBalancedHost tbh = timelyHosts.get(r.nextInt(numHosts));
+        LOG.debug("Unregistering:" + TimelyBalancedHost.toStringShort(tbh));
+        ServiceRegistry.unregisterService(curatorFramework, tbh);
+        return tbh;
+    }
+
+    @Test
+    public void test1() {
+        LOG.debug("Configure test");
+        int numHosts = 10;
+        BalancerConfiguration balancerConfiguration = getBalancerConfiguration();
+        createTimelyHosts(balancerConfiguration, numHosts);
+        TestHealthChecker healthChecker = new TestHealthChecker(balancerConfiguration, timelyHosts);
+        BalancedMetricResolver balancedMetricResolver = new BalancedMetricResolver(curatorFramework,
+                balancerConfiguration, healthChecker);
+        registerCallbacks(balancedMetricResolver);
+        LOG.debug("Start balancedMetricResolver");
+        balancedMetricResolver.start();
+        long now = System.currentTimeMillis();
+        int numMetrics = 100;
+        for (int x = 0; x < numMetrics; x++) {
+            balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", x));
+        }
+
+        LOG.debug("Begin monitoring");
+        long cycle = 0;
+        List<TimelyBalancedHost> serversUnregistered = new ArrayList<>();
+        while (System.currentTimeMillis() < now + 15000) {
+            try {
+                Thread.sleep(100);
+                for (int x = 0; x < 10000; x++) {
+                    int mNum = r.nextInt(numMetrics);
+                    balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", mNum));
+                }
+                // if (cycle % 100 == 0) {
+                // randomServerDown(healthChecker, numHosts);
+                // }
+                // if (cycle % 2 == 0) {
+                // randomServerToggle(healthChecker, numHosts);
+                // }
+                if (cycle % 100 == 0) {
+                    serversUnregistered.add(randomServerUnregister(numHosts));
+                }
+                if (cycle % 500 == 0) {
+                    serversUnregistered.stream().forEach(tbh -> ServiceRegistry.registerService(curatorFramework, tbh));
+                    serversUnregistered.clear();
+                }
+
+            } catch (InterruptedException e) {
+
+            }
+            cycle++;
+        }
+        LOG.debug("Stop balancedMetricResolver");
+        balancedMetricResolver.stop();
+    }
+
+    @Test
+    public void test2() {
+        LOG.debug("Configure test");
+        int numHosts = 10;
+        BalancerConfiguration balancerConfiguration = getBalancerConfiguration();
+        createTimelyHosts(balancerConfiguration, numHosts);
+        TestHealthChecker healthChecker = new TestHealthChecker(balancerConfiguration, timelyHosts);
+        BalancedMetricResolver balancedMetricResolver = new BalancedMetricResolver(curatorFramework,
+                balancerConfiguration, healthChecker);
+        registerCallbacks(balancedMetricResolver);
+        LOG.debug("Start balancedMetricResolver");
+        balancedMetricResolver.start();
+        long now = System.currentTimeMillis();
+        int numMetrics = 100;
+        for (int x = 0; x < numMetrics; x++) {
+            balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", x));
+        }
+
+        LOG.debug("Begin monitoring");
+        long cycle = 0;
+        List<TimelyBalancedHost> serversUnregistered = new ArrayList<>();
+        while (System.currentTimeMillis() < now + 15000) {
+            try {
+                Thread.sleep(100);
+                for (int x = 0; x < 10000; x++) {
+                    int mNum = r.nextInt(numMetrics);
+                    balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", mNum));
+                }
+                // if (cycle % 100 == 0) {
+                // randomServerDown(healthChecker, numHosts);
+                // }
+                // if (cycle % 2 == 0) {
+                // randomServerToggle(healthChecker, numHosts);
+                // }
+                if (cycle % 100 == 0) {
+                    serversUnregistered.add(randomServerUnregister(numHosts));
+                }
+                if (cycle % 500 == 0) {
+                    serversUnregistered.stream().forEach(tbh -> ServiceRegistry.registerService(curatorFramework, tbh));
+                    serversUnregistered.clear();
+                }
+
+            } catch (InterruptedException e) {
+
+            }
+            cycle++;
+        }
+        LOG.debug("Stop balancedMetricResolver");
+        balancedMetricResolver.stop();
+    }
+}

balancer/src/test/java/timely/balancer/resolver/TestHealthChecker.java

@@ -0,0 +1,89 @@
+package timely.balancer.resolver;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.balancer.configuration.BalancerConfiguration;
+import timely.balancer.connection.TimelyBalancedHost;
+import timely.balancer.healthcheck.HealthChecker;
+import timely.balancer.healthcheck.TcpHealthChecker;
+
+public class TestHealthChecker implements HealthChecker {
+
+    private final Logger LOG = LoggerFactory.getLogger(TcpHealthChecker.class);
+    private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
+    private Map<TimelyBalancedHost, Boolean> serverUpMap = Collections.synchronizedMap(new HashMap<>());
+    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
+
+    public TestHealthChecker(BalancerConfiguration config, List<TimelyBalancedHost> timelyHosts) {
+        setTimelyHosts(timelyHosts);
+        this.executorService.scheduleAtFixedRate(() -> {
+            synchronized (timelyHosts) {
+                for (TimelyBalancedHost h : timelyHosts) {
+                    try {
+                        if (TestHealthChecker.this.isServerHealthy(h)) {
+                            h.reportSuccess();
+                        } else {
+                            h.reportFailure();
+                        }
+                    } catch (Exception e) {
+                        LOG.error(e.getMessage(), e);
+                        h.reportFailure();
+                    }
+                }
+            }
+        }, 0, config.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
+        synchronized (timelyHosts) {
+            this.timelyHosts.clear();
+            this.timelyHosts.addAll(timelyHosts);
+            for (TimelyBalancedHost h : timelyHosts) {
+                serverUpMap.put(h, true);
+            }
+        }
+
+    }
+
+    public boolean isServerHealthy(TimelyBalancedHost h) {
+        return serverUpMap.get(h);
+    }
+
+    public void serverUp(TimelyBalancedHost h) {
+        serverUpMap.put(h, true);
+    }
+
+    public void serverDown(TimelyBalancedHost h) {
+        serverUpMap.put(h, false);
+    }
+
+    public void serverToggle(TimelyBalancedHost h) {
+        synchronized (serverUpMap) {
+            serverUpMap.put(h, !serverUpMap.get(h));
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        this.executorService.shutdown();
+        try {
+            this.executorService.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.executorService.isTerminated()) {
+                this.executorService.shutdownNow();
+            }
+        }
+    }
+}

balancer/src/test/resources/log4j2.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+  <Appenders>
+    <Console name="CONSOLE" target="SYSTEM_OUT">
+      <PatternLayout pattern="%d{yyyyMMdd HHmmss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+    </Console>
+  </Appenders>
+  <Loggers>
+    <Logger name="org.apache.accumulo.core.client.impl.ThriftTransportPool" level="OFF" additivity="false" />
+    <Logger name="org.apache.accumulo.fate.zookeeper.ZooCache" level="OFF" additivity="false" />
+    <Logger name="org.apache.zookeeper" level="OFF" additivity="false" />
+    <Logger name="timely.balancer.resolver" level="TRACE" additivity="true" />
+    <Logger name="timely.balancer.resolver.TestBalancedMetricResolver" level="DEBUG" additivity="true" />
+    <Root level="ERROR">
+      <AppenderRef ref="CONSOLE"/>
+    </Root>
+  </Loggers>
+</Configuration>

grafana-auth/src/main/java/timely/grafana/auth/GrafanaAuth.java

@@ -81,8 +81,11 @@ public class GrafanaAuth {
         this.grafanaAuthConfig = grafanaAuthConfig;
     }
 
-    protected static ConfigurableApplicationContext initializeConfiguration(String[] args) {
+    protected static ConfigurableApplicationContext initializeContext(String[] args) {
-        return new SpringApplicationBuilder(SpringBootstrap.class).web(WebApplicationType.NONE).run(args);
+        SpringApplicationBuilder builder = new SpringApplicationBuilder(SpringBootstrap.class);
+        builder.web(WebApplicationType.NONE);
+        builder.registerShutdownHook(false);
+        return builder.run(args);
     }
 
     private void shutdownHook() {
@@ -109,13 +112,11 @@ public class GrafanaAuth {
             try {
                 f.get();
             } catch (final Exception e) {
-                LOG.error("Error while shutting down channel: {}", f.channel().config());
+                LOG.error("Channel:" + f.channel().config() + " -> " + e.getMessage(), e);
-                LOG.error("{}", e.getMessage());
-                e.printStackTrace();
             }
         });
 
-        Integer quietPeriod = grafanaAuthConfig.getShutdownQuietPeriod();
+        int quietPeriod = grafanaAuthConfig.getShutdownQuietPeriod();
         List<Future<?>> groupFutures = new ArrayList<>();
 
         if (httpBossGroup != null) {
@@ -132,9 +133,7 @@ public class GrafanaAuth {
             try {
                 f.get();
             } catch (final Exception e) {
-                LOG.error("Error while shutting down group: {}", f.toString());
+                LOG.error("Group:" + f.toString() + " -> " + e.getMessage(), e);
-                LOG.error("{}", e.getMessage());
-                e.printStackTrace();
             }
         });
 
@@ -272,12 +271,13 @@ public class GrafanaAuth {
 
     public static void main(String[] args) throws Exception {
 
-        GrafanaAuth.applicationContext = GrafanaAuth.initializeConfiguration(args);
+        GrafanaAuth.applicationContext = GrafanaAuth.initializeContext(args);
-        GrafanaAuthConfiguration balancerConf = applicationContext.getBean(GrafanaAuthConfiguration.class);
+        GrafanaAuthConfiguration grafanaAuthConf = GrafanaAuth.applicationContext
+                .getBean(GrafanaAuthConfiguration.class);
 
-        GrafanaAuth b = new GrafanaAuth(balancerConf);
+        GrafanaAuth grafanaAuth = new GrafanaAuth(grafanaAuthConf);
         try {
-            b.run();
+            grafanaAuth.run();
             LATCH.await();
         } catch (final InterruptedException e) {
             LOG.info("Server shutting down.");
@@ -285,7 +285,7 @@ public class GrafanaAuth {
             LOG.error("Error running server.", e);
         } finally {
             try {
-                b.shutdown();
+                grafanaAuth.shutdown();
             } catch (Exception e) {
                 System.exit(1);
             }
@@ -314,7 +314,8 @@ public class GrafanaAuth {
                 ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler());
                 ch.pipeline().addLast("login", new X509LoginRequestHandler(config.getSecurity(), config.getHttp()));
                 ch.pipeline().addLast("httpRelay", new GrafanaRelayHandler(config, httpClientPool));
-                ch.pipeline().addLast("error", new TimelyExceptionHandler());
+                ch.pipeline().addLast("error", new TimelyExceptionHandler()
+                        .setIgnoreSslHandshakeErrors(config.getSecurity().getServerSsl().isUseGeneratedKeypair()));
             }
         };
     }

grafana-auth/src/main/java/timely/grafana/auth/netty/http/GrafanaRelayHandler.java

@@ -87,7 +87,7 @@ public class GrafanaRelayHandler extends SimpleChannelInboundHandler<HttpRequest
                 String protocol = config.getGrafana().getProtocol();
                 String host = config.getGrafana().getHost();
                 Integer port = config.getGrafana().getPort();
-                k = TimelyBalancedHost.of(host, -1, port, -1, -1);
+                k = TimelyBalancedHost.of(host, -1, port, -1, -1, null);
                 client = httpClientPool.borrowObject(k);
                 Multimap<String, String> headers = HttpHeaderUtils.toMultimap(request.headers());
 

server/pom.xml

@@ -108,6 +108,10 @@
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-minicluster</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.glassfish.tyrus</groupId>
+      <artifactId>tyrus-client</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>

server/src/main/java/timely/Server.java

@@ -76,7 +76,6 @@ import org.slf4j.LoggerFactory;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.context.ConfigurableApplicationContext;
-import timely.api.response.TimelyException;
 import timely.auth.AuthCache;
 import timely.auth.VisibilityCache;
 import timely.configuration.Configuration;
@@ -234,11 +233,12 @@ public class Server {
 
     public static void main(String[] args) throws Exception {
 
-        Configuration conf = initializeConfiguration(args);
+        Server.applicationContext = initializeContext(args);
+        Configuration conf = Server.applicationContext.getBean(Configuration.class);
 
-        Server s = new Server(conf);
+        Server server = new Server(conf);
         try {
-            s.run();
+            server.run();
             LATCH.await();
         } catch (final InterruptedException e) {
             LOG.info("Server shutting down.");
@@ -246,16 +246,18 @@ public class Server {
             LOG.error("Error running server.", e);
         } finally {
             try {
-                s.shutdown();
+                server.shutdown();
             } catch (Exception e) {
                 System.exit(1);
             }
         }
     }
 
-    protected static Configuration initializeConfiguration(String[] args) {
+    protected static ConfigurableApplicationContext initializeContext(String[] args) {
-        applicationContext = new SpringApplicationBuilder(SpringBootstrap.class).web(WebApplicationType.NONE).run(args);
+        SpringApplicationBuilder builder = new SpringApplicationBuilder(SpringBootstrap.class);
-        return applicationContext.getBean(Configuration.class);
+        builder.web(WebApplicationType.NONE);
+        builder.registerShutdownHook(false);
+        return builder.run(args);
     }
 
     private void shutdownHook() {
@@ -270,6 +272,8 @@ public class Server {
     }
 
     public void shutdown() {
+        LOG.warn("Server shutdown 1");
+
         List<ChannelFuture> channelFutures = new ArrayList<>();
 
         if (tcpChannelHandle != null) {
@@ -287,8 +291,9 @@ public class Server {
             channelFutures.add(wsChannelHandle.close());
         }
 
+        int udpChannel = 1;
         for (Channel c : udpChannelHandleList) {
-            LOG.info("Closing udpChannelHandle");
+            LOG.info("Closing udpChannelHandle #" + udpChannel++);
             channelFutures.add(c.close());
         }
 
@@ -297,14 +302,15 @@ public class Server {
             try {
                 f.get();
             } catch (final Exception e) {
-                LOG.error("Error while shutting down channel: {}", f.channel().config());
+                LOG.error("Channel:" + f.channel().config() + " -> " + e.getMessage(), e);
-                LOG.error("{}", e.getMessage());
-                e.printStackTrace();
             }
         });
 
-        Integer quietPeriod = config.getServer().getShutdownQuietPeriod();
+        LOG.warn("Server shutdown 2");
+
+        int quietPeriod = config.getServer().getShutdownQuietPeriod();
         List<Future<?>> groupFutures = new ArrayList<>();
+
         if (tcpBossGroup != null) {
             LOG.info("Shutting down tcpBossGroup");
             groupFutures.add(tcpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
@@ -340,6 +346,8 @@ public class Server {
             groupFutures.add(udpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
+        LOG.warn("Server shutdown 3");
+
         if (udpWorkerGroup != null) {
             LOG.info("Shutting down udpWorkerGroup");
             groupFutures.add(udpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
@@ -349,41 +357,70 @@ public class Server {
             try {
                 f.get();
             } catch (final Exception e) {
-                LOG.error("Error while shutting down group: {}", f.toString());
+                LOG.error("Group:" + f.toString() + " -> " + e.getMessage(), e);
-                LOG.error("{}", e.getMessage());
-                e.printStackTrace();
             }
         });
 
-        try {
+        LOG.warn("Server shutdown 4");
-            LOG.info("Flushing datastore.");
-            dataStore.flush();
-        } catch (TimelyException e) {
-            LOG.error("Error flushing to server during shutdown", e);
-        }
 
         try {
-            LOG.info("Closing MetaCacheFactory");
+            LOG.info("Closing dataStore");
+            dataStore.close();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+
+        LOG.warn("Server shutdown 5");
+
+        if (dataStoreCache != null) {
+            try {
+                LOG.info("Closing dataStoreCache");
+                dataStoreCache.close();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+
+        LOG.warn("Server shutdown 6");
+
+        try {
+            LOG.info("Closing metaCacheFactory");
             MetaCacheFactory.close();
         } catch (Exception e) {
-            LOG.error("Error closing MetaCacheFactory during shutdown", e);
+            LOG.error(e.getMessage(), e);
         }
 
+        LOG.warn("Server shutdown 7");
+
         try {
-            LOG.info("Closing WebSocketRequestDecoder");
+            LOG.info("Closing webSocketRequestDecoder subscriptions");
             WebSocketRequestDecoder.close();
         } catch (Exception e) {
-            LOG.error("Error closing WebSocketRequestDecoder during shutdown", e);
+            LOG.error(e.getMessage(), e);
         }
 
+        LOG.warn("Server shutdown 8");
+
         if (curatorFramework != null) {
-            curatorFramework.close();
+            try {
+                LOG.info("Closing curatorFramework");
+                curatorFramework.close();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
         }
 
+        LOG.warn("Server shutdown 9");
+
         if (applicationContext != null) {
-            LOG.info("Closing applicationContext");
+            try {
-            applicationContext.close();
+                LOG.info("Closing applicationContext");
+                applicationContext.close();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
         }
+        LOG.warn("Server shutdown 10");
         this.shutdown = true;
         LOG.info("Server shut down.");
     }
@@ -547,7 +584,7 @@ public class Server {
         if (generate) {
             LOG.warn("Using generated self signed server certificate");
             Date begin = new Date();
-            Date end = new Date(begin.getTime() + 86400000);
+            Date end = new Date(begin.getTime() + TimeUnit.DAYS.toMillis(7));
             SelfSignedCertificate ssc = new SelfSignedCertificate("localhost", begin, end);
             ssl = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey());
         } else {
@@ -614,7 +651,8 @@ public class Server {
                 ch.pipeline().addLast("cors", new CorsHandler(cors));
                 ch.pipeline().addLast("queryDecoder",
                         new timely.netty.http.HttpRequestDecoder(config.getSecurity(), config.getHttp()));
-                ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler());
+                ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler()
+                        .setIgnoreSslHandshakeErrors(config.getSecurity().getServerSsl().isUseGeneratedKeypair()));
                 ch.pipeline().addLast("strict", new StrictTransportHandler(config));
                 ch.pipeline().addLast("login", new X509LoginRequestHandler(config.getSecurity(), config.getHttp()));
                 ch.pipeline().addLast("doLogin",
@@ -627,7 +665,9 @@ public class Server {
                 ch.pipeline().addLast("version", new HttpVersionRequestHandler());
                 ch.pipeline().addLast("cache", new HttpCacheRequestHandler(dataStoreCache));
                 ch.pipeline().addLast("put", new HttpMetricPutHandler(dataStore));
-                ch.pipeline().addLast("error", new TimelyExceptionHandler());
+                ch.pipeline().addLast("error", new TimelyExceptionHandler()
+                        .setIgnoreSslHandshakeErrors(config.getSecurity().getServerSsl().isUseGeneratedKeypair()));
+
             }
         };
     }

server/src/main/java/timely/StandaloneServer.java

@@ -52,7 +52,8 @@ public class StandaloneServer extends Server {
             System.exit(1);
         }
 
-        Configuration conf = initializeConfiguration(args);
+        StandaloneServer.applicationContext = initializeContext(args);
+        Configuration conf = StandaloneServer.applicationContext.getBean(Configuration.class);
 
         File accumuloDir = new File(tmp, "accumulo");
         MiniAccumuloConfig macConfig = new MiniAccumuloConfig(accumuloDir, "secret");

server/src/main/java/timely/netty/http/HttpStaticFileServerHandler.java

@@ -48,6 +48,7 @@ import java.util.TimeZone;
 import java.util.regex.Pattern;
 
 import javax.activation.MimetypesFileTypeMap;
+import javax.net.ssl.SSLHandshakeException;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
@@ -123,6 +124,7 @@ public class HttpStaticFileServerHandler extends SimpleChannelInboundHandler<Ful
     public static final String HTTP_DATE_FORMAT = "EEE, dd MMM yyyy HH:mm:ss zzz";
     public static final String HTTP_DATE_GMT_TIMEZONE = "GMT";
     public static final int HTTP_CACHE_SECONDS = 60;
+    private boolean ignoreSslHandshakeErrors = false;
 
     @Override
     public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
@@ -221,6 +223,10 @@ public class HttpStaticFileServerHandler extends SimpleChannelInboundHandler<Ful
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        // ignore SSLHandshakeException when using a self-signed server certificate
+        if (ignoreSslHandshakeErrors && cause.getCause() instanceof SSLHandshakeException) {
+            return;
+        }
         LOG.error("Error serving file", cause);
         if (ctx.channel().isActive()) {
             sendError(ctx, INTERNAL_SERVER_ERROR);
@@ -379,4 +385,9 @@ public class HttpStaticFileServerHandler extends SimpleChannelInboundHandler<Ful
         MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap();
         response.headers().set(CONTENT_TYPE, mimeTypesMap.getContentType(file.getPath()));
     }
+
+    public HttpStaticFileServerHandler setIgnoreSslHandshakeErrors(boolean ignoreSslHandshakeErrors) {
+        this.ignoreSslHandshakeErrors = ignoreSslHandshakeErrors;
+        return this;
+    }
 }

server/src/main/java/timely/netty/http/TimelyExceptionHandler.java

@@ -1,5 +1,7 @@
 package timely.netty.http;
 
+import javax.net.ssl.SSLHandshakeException;
+
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
@@ -12,6 +14,7 @@ import timely.api.response.TimelyException;
 public class TimelyExceptionHandler extends SimpleChannelInboundHandler<TimelyException> implements TimelyHttpHandler {
 
     private static final Logger LOG = LoggerFactory.getLogger(TimelyExceptionHandler.class);
+    private boolean ignoreSslHandshakeErrors = false;
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, TimelyException msg) throws Exception {
@@ -20,6 +23,10 @@ public class TimelyExceptionHandler extends SimpleChannelInboundHandler<TimelyEx
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        // ignore SSLHandshakeException when using a self-signed server certificate
+        if (ignoreSslHandshakeErrors && cause.getCause() instanceof SSLHandshakeException) {
+            return;
+        }
         LOG.error("Unhandled exception in pipeline", cause);
         if (cause instanceof TimelyException) {
             this.sendHttpError(ctx, (TimelyException) cause);
@@ -32,4 +39,8 @@ public class TimelyExceptionHandler extends SimpleChannelInboundHandler<TimelyEx
         }
     }
 
+    public TimelyExceptionHandler setIgnoreSslHandshakeErrors(boolean ignoreSslHandshakeErrors) {
+        this.ignoreSslHandshakeErrors = ignoreSslHandshakeErrors;
+        return this;
+    }
 }

server/src/main/java/timely/netty/websocket/WebSocketRequestDecoder.java

@@ -109,7 +109,6 @@ public class WebSocketRequestDecoder extends MessageToMessageDecoder<WebSocketFr
     }
 
     public static void close() {
-        LOG.info("Closing subscriptions");
         SubscriptionRegistry.get().forEach((k, v) -> v.close());
     }
 

server/src/main/java/timely/store/DataStore.java

@@ -29,8 +29,6 @@ public interface DataStore {
 
     List<QueryResponse> query(QueryRequest msg) throws TimelyException;
 
-    void flush() throws TimelyException;
-
     long getAgeOffForMetric(String metricName);
 
     List<Range> getQueryRanges(String metric, long start, long end, Set<Tag> colFamValues);
@@ -45,4 +43,6 @@ public interface DataStore {
     DataStoreCache getCache();
 
     InternalMetrics getInternalMetrics();
+
+    void close() throws Exception;
 }

server/src/main/java/timely/store/DataStoreImpl.java

@@ -22,10 +22,10 @@ import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
 import java.util.SortedMap;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -127,7 +127,7 @@ public class DataStoreImpl implements DataStore {
     private final String metricsTable;
     private final String metaTable;
     private final InternalMetrics internalMetrics;
-    private final Timer internalMetricsTimer = new Timer("InternalMetricsTimer", true);
+    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
     private final int scannerThreads;
     private final long maxDownsampleMemory;
     private final BatchWriterConfig bwConfig;
@@ -244,14 +244,9 @@ public class DataStoreImpl implements DataStore {
             }
 
             internalMetrics = new InternalMetrics(conf);
-            internalMetricsTimer.schedule(new TimerTask() {
+            executorService.scheduleAtFixedRate(
-
+                    () -> internalMetrics.getMetricsAndReset().forEach(m -> store(m, false)), METRICS_PERIOD,
-                @Override
+                    METRICS_PERIOD, TimeUnit.MILLISECONDS);
-                public void run() {
-                    internalMetrics.getMetricsAndReset().forEach(m -> store(m, false));
-                }
-
-            }, METRICS_PERIOD, METRICS_PERIOD);
 
             this.metaCache = MetaCacheFactory.getCache(conf);
         } catch (Exception e2) {
@@ -484,19 +479,6 @@ public class DataStoreImpl implements DataStore {
         return result;
     }
 
-    @Override
-    public void flush() {
-        internalMetricsTimer.cancel();
-        writers.forEach(w -> {
-            try {
-                w.close();
-            } catch (final Exception ex) {
-                LOG.warn("Error shutting down batchwriter", ex);
-            }
-
-        });
-    }
-
     @Override
     public SearchLookupResponse lookup(SearchLookupRequest msg) throws TimelyException {
         long startMillis = System.currentTimeMillis();
@@ -1054,4 +1036,24 @@ public class DataStoreImpl implements DataStore {
         }
     }
 
+    @Override
+    public void close() throws Exception {
+        this.executorService.shutdown();
+        try {
+            this.executorService.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.executorService.isTerminated()) {
+                this.executorService.shutdownNow();
+            }
+        }
+        writers.forEach(w -> {
+            try {
+                w.close();
+            } catch (final Exception ex) {
+                LOG.warn("Error shutting down batchwriter: ", ex.getMessage());
+            }
+        });
+    }
 }

server/src/main/java/timely/store/MetaCacheImpl.java

@@ -5,9 +5,10 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.TreeSet;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
@@ -37,7 +38,7 @@ public class MetaCacheImpl implements MetaCache {
     private volatile boolean closed = false;
     private Cache<Meta, Object> cache = null;
     private Configuration configuration;
-    private Timer timer = new Timer("MetaCacheImpl", true);
+    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
 
     @Override
     public void init(Configuration config) {
@@ -45,13 +46,8 @@ public class MetaCacheImpl implements MetaCache {
         cache = Caffeine.newBuilder().initialCapacity(1000).build();
         long cacheRefreshMinutes = configuration.getMetaCache().getCacheRefreshMinutes();
         if (cacheRefreshMinutes > 0) {
-            timer.schedule(new TimerTask() {
+            executorService.scheduleAtFixedRate(() -> refreshCache(config.getMetaCache().getExpirationMinutes()), 1,
-
+                    cacheRefreshMinutes, TimeUnit.MINUTES);
-                @Override
-                public void run() {
-                    refreshCache(config.getMetaCache().getExpirationMinutes());
-                }
-            }, 60000, cacheRefreshMinutes * 60 * 1000);
         }
     }
 
@@ -72,7 +68,10 @@ public class MetaCacheImpl implements MetaCache {
                     new PasswordToken(accumuloConf.getPassword()));
             scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
             LOG.debug("Begin scanning " + metaTable);
-            Range metricRange = new Range(new Key(Meta.METRIC_PREFIX), true, new Key(Meta.METRIC_PREFIX + '\0'), false);
+            Key metricPrefixBeginKey = new Key(Meta.METRIC_PREFIX);
+            int firstChar = Meta.METRIC_PREFIX.charAt(0);
+            Key metricPrefixEndKey = new Key(String.valueOf((char) (firstChar + 1)) + ':');
+            Range metricRange = new Range(metricPrefixBeginKey, true, metricPrefixEndKey, false);
             scanner.setRange(metricRange);
             Set<String> allMetrics = new TreeSet<>();
             for (Map.Entry<Key, Value> entry : scanner) {
@@ -145,6 +144,16 @@ public class MetaCacheImpl implements MetaCache {
     @Override
     public void close() {
         this.closed = true;
+        this.executorService.shutdown();
+        try {
+            this.executorService.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.executorService.isTerminated()) {
+                this.executorService.shutdownNow();
+            }
+        }
     }
 
     @Override

server/src/main/java/timely/store/cache/DataStoreCache.java

@@ -11,9 +11,9 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.TreeSet;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.StampedLock;
 
@@ -79,7 +79,7 @@ public class DataStoreCache {
     private int flushBatch = 0;
     private int numBatches = 5;
     private InternalMetrics internalMetrics;
-    private Timer maintenanceTimer = new Timer("DataStoreCacheTimer");
+    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(6);
 
     public DataStoreCache(CuratorFramework curatorFramework, Configuration conf) {
         if (curatorFramework != null) {
@@ -103,82 +103,58 @@ public class DataStoreCache {
 
         minimumAgeOffForIterator = getAgeOffForIterator(minimumAgeOff);
 
-        maintenanceTimer.schedule(new TimerTask() {
+        executorService.scheduleAtFixedRate(() -> {
-
+            try {
-            @Override
+                flushCaches(flushBatch);
-            public void run() {
+            } catch (Exception e) {
-                try {
+                LOG.error(e.getMessage(), e);
-                    flushCaches(flushBatch);
+            } finally {
-                } catch (Exception e) {
+                flushBatch++;
-                    LOG.error(e.getMessage(), e);
+                if (flushBatch == numBatches) {
-                } finally {
+                    flushBatch = 0;
-                    flushBatch++;
-                    if (flushBatch == numBatches) {
-                        flushBatch = 0;
-                    }
                 }
             }
-        }, conf.getCache().getFlushInterval(), conf.getCache().getFlushInterval());
+        }, conf.getCache().getFlushInterval(), conf.getCache().getFlushInterval(), TimeUnit.MILLISECONDS);
 
-        maintenanceTimer.schedule(new TimerTask() {
+        executorService.scheduleAtFixedRate(() -> {
-
+            try {
-            @Override
+                DataStoreCache.this.removeStaleMetrics();
-            public void run() {
+            } catch (Exception e) {
-                try {
+                LOG.error(e.getMessage(), e);
-                    removeStaleMetrics();
-                } catch (Exception e) {
-                    LOG.error(e.getMessage(), e);
-                }
             }
-        }, (300 * 1000), (300 * 1000));
+        }, 5, 5, TimeUnit.MINUTES);
 
-        maintenanceTimer.schedule(new TimerTask() {
+        executorService.scheduleAtFixedRate(() -> {
-
+            try {
-            @Override
+                ageOffGorillaStores();
-            public void run() {
+            } catch (Exception e) {
-                try {
+                LOG.error(e.getMessage(), e);
-                    ageOffGorillaStores();
-                } catch (Exception e) {
-                    LOG.error(e.getMessage(), e);
-                }
             }
-        }, (600 * 1000), (600 * 1000));
+        }, 10, 10, TimeUnit.MINUTES);
 
-        maintenanceTimer.schedule(new TimerTask() {
+        executorService.scheduleAtFixedRate(() -> {
-
+            try {
-            @Override
+                archiveGorillaStoreCurrentCompressors();
-            public void run() {
+            } catch (Exception e) {
-                try {
+                LOG.error(e.getMessage(), e);
-                    archiveGorillaStoreCurrentCompressors();
-                } catch (Exception e) {
-                    LOG.error(e.getMessage(), e);
-                }
             }
-        }, (3600 * 1000), (3600 * 1000));
+        }, 60, 60, TimeUnit.MINUTES);
 
-        maintenanceTimer.schedule(new TimerTask() {
+        executorService.scheduleAtFixedRate(() -> {
-
+            try {
-            @Override
+                pruneStats();
-            public void run() {
+            } catch (Exception e) {
-                try {
+                LOG.error(e.getMessage(), e);
-                    pruneStats();
-                } catch (Exception e) {
-                    LOG.error(e.getMessage(), e);
-                }
             }
-        }, (600 * 1000), (600 * 1000));
+        }, 10, 10, TimeUnit.MINUTES);
 
-        maintenanceTimer.schedule(new TimerTask() {
+        executorService.scheduleAtFixedRate(() -> {
-
+            try {
-            @Override
+                reportInternalMetrics();
-            public void run() {
+            } catch (Exception e) {
-                try {
+                LOG.error(e.getMessage(), e);
-                    reportInternalMetrics();
-                } catch (Exception e) {
-                    LOG.error(e.getMessage(), e);
-                }
             }
-        }, (60 * 1000), (60 * 1000));
+        }, 1, 1, TimeUnit.MINUTES);
     }
 
     private void addNonCachedMetrics(Collection<String> nonCachedMetricsUpdate) {
@@ -230,6 +206,11 @@ public class DataStoreCache {
         nonCachedMetricsIPRWLock = new InterProcessReadWriteLock(curatorFramework, NON_CACHED_METRICS_LOCK_PATH);
         testIPRWLock(curatorFramework, nonCachedMetricsIPRWLock, NON_CACHED_METRICS_LOCK_PATH);
         nonCachedMetricsIP = new DistributedAtomicValue(curatorFramework, NON_CACHED_METRICS, new RetryForever(1000));
+        try {
+            nonCachedMetricsIP.initialize(SerializationUtils.serialize(new TreeSet<>()));
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
         TreeCacheListener nonCachedMetricsListener = new TreeCacheListener() {
 
             @Override
@@ -930,4 +911,17 @@ public class DataStoreCache {
         }
         return response;
     }
+
+    public void close() {
+        this.executorService.shutdown();
+        try {
+            this.executorService.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.executorService.isTerminated()) {
+                this.executorService.shutdownNow();
+            }
+        }
+    }
 }